From 8e0948a0ce0f27cb679b7aa8fe94ff0fb14b330e Mon Sep 17 00:00:00 2001 From: Mitchell Hashimoto Date: Fri, 01 May 2026 22:40:33 +0000 Subject: [PATCH] provider router --- README.md | 7 ++++--- main.go | 27 ++++++++++++++------------- provider_router.go | 149 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ provider_router_test.go | 229 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 4 file(s) changed, 396 insertion(s)(+), 16 deletion(s)(-) diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -80,9 +80,10 @@ | `TACK_DB_PATH` | Local SQLite path (default `tack.db`) | | `TACK_JETSTREAM_URL` | Tangled Jetstream WebSocket URL | | `TACK_DEV` | Use `ws://` for knot event-streams (any non-empty value) | -When no provider is configured, tack runs an in-process fake provider -that's useful for exercising the jetstream → knot → `/events` flow -locally without a real CI account. +All configured providers are active simultaneously. Each workflow +chooses its provider via the first key under its top-level `tack:` +block. e.g. `tack: { buildkite: { ... } }` runs on Buildkite, +`tack: { fake: {} }` runs on the in-process fake provider. ## Providers diff --git a/main.go b/main.go --- a/main.go +++ b/main.go @@ -159,20 +159,23 @@ // cursor. Constructed before the consumers in case we ever want // them to publish synthetic status events at startup. br := newBroker(st) - // Provider that turns Tangled pipeline triggers into - // pipeline.status events. The Buildkite provider is the real - // integration; the fake one stands in when no Buildkite token is - // configured so the full jetstream → knot → /events flow is - // still exercisable locally without a Buildkite account. + // Providers turn Tangled pipeline triggers into pipeline.status + // events. We always wire the fake provider in so workflows can + // opt into it (via `tack: { fake: ... }`) for end-to-end testing + // even when Buildkite credentials are present; the Buildkite + // provider is added on top when configured. Routing is per- + // workflow and driven by the workflow YAML — see providerRouter. // // bkProvider is kept as a typed pointer separately because the // /webhooks/buildkite handler needs the concrete *buildkiteProvider // (for HandleWebhook + signature verification), not the abstract // Provider surface. - var ( - provider Provider - bkProvider *buildkiteProvider - ) + providers := map[string]Provider{ + "fake": newFakeProvider(br, logger), + } + logger.Info("fake provider enabled (workflow opt-in via `tack.fake:`)") + + var bkProvider *buildkiteProvider if cfg.BuildkiteToken != "" { bkProvider = newBuildkiteProvider( br, st, @@ -182,15 +185,13 @@ cfg.BuildkiteWebhookSecret, cfg.BuildkiteWebhookMode, logger, ) - provider = bkProvider + providers["buildkite"] = bkProvider logger.Info("buildkite provider enabled", "default_org", cfg.BuildkiteOrg, "webhook_mode", cfg.BuildkiteWebhookMode, ) - } else { - provider = newFakeProvider(br, logger) - logger.Info("fake provider enabled (set TACK_BUILDKITE_TOKEN to use buildkite)") } + provider := newProviderRouter(logger, providers) // Start the knot event-stream consumer first so the jetstream // loop has somewhere to register newly-observed knots into. It diff --git a/provider_router.go b/provider_router.go new file mode 100644 --- /dev/null +++ b/provider_router.go @@ -0,0 +1,149 @@ +package main + +// providerRouter dispatches each incoming workflow to whichever +// configured Provider matches the workflow's YAML body. Selection +// happens per-workflow: we decode the top-level `tack:` map and pick +// the first child key that names one of the registered providers. +// This keeps the trigger plumbing oblivious to which backend any +// given workflow will run on, and lets a single tack instance host +// multiple providers concurrently — the workflow YAML is the source +// of truth for routing, not the operator's env. +// +// Providers are registered as a (key → Provider) map. When a +// workflow names more than one provider key under `tack:` (a config +// mistake in practice) the router walks the YAML's child keys in +// document order and picks the first one that has a registered +// provider — Go's map iteration randomness never enters into it, +// because the YAML's MapSlice preserves order. + +import ( + "context" + "errors" + "fmt" + "log/slog" + "strings" + + "go.yaml.in/yaml/v2" + "tangled.org/core/api/tangled" +) + +// providerRouter is itself a Provider so the rest of tack (knot +// consumer, HTTP handlers) keeps talking to a single Provider value +// regardless of how many real backends are wired in. +type providerRouter struct { + log *slog.Logger + providers map[string]Provider +} + +// Compile-time interface conformance check. +var _ Provider = (*providerRouter)(nil) + +// newProviderRouter wires a router from a (key → Provider) map. +func newProviderRouter( + log *slog.Logger, + providers map[string]Provider, +) *providerRouter { + return &providerRouter{ + log: log.With("component", "provider", "kind", "router"), + providers: providers, + } +} + +// Spawn satisfies Provider. Each workflow is routed independently — +// a single pipeline can mix workflows that target different +// providers. Workflows whose YAML doesn't name a known provider are +// logged loudly and skipped, since silently dropping them would +// hide a config error from the operator. +func (r *providerRouter) Spawn( + ctx context.Context, + knot string, + pipelineRkey string, + trigger *tangled.Pipeline_TriggerMetadata, + workflows []*tangled.Pipeline_Workflow, +) { + for _, wf := range workflows { + // Defensive: the lexicon allows nil entries and doesn't + // require a name. We can't route a workflow that has no + // body to inspect. + if wf == nil || wf.Name == "" { + continue + } + p, err := r.pick(wf.Raw) + if err != nil { + r.log.Error("route workflow", + "err", err, + "knot", knot, + "pipeline_rkey", pipelineRkey, + "workflow", wf.Name, + ) + continue + } + // Hand the workflow off as a single-element slice so the + // downstream provider's existing Spawn loop runs unchanged. + p.Spawn(ctx, knot, pipelineRkey, trigger, + []*tangled.Pipeline_Workflow{wf}, + ) + } +} + +// Logs satisfies Provider. The (knot, rkey, workflow) tuple alone +// doesn't tell us which backend ran the workflow — the YAML body +// isn't carried on the request — so we ask each provider and +// surface the first one that has a stream. ErrLogsNotFound from a +// provider just means "not mine"; we keep walking. Any other error +// is the answer (a real backend failure should surface to the HTTP +// caller, not be masked by the next provider). +// +// Map iteration order is undefined, but in practice exactly one +// provider should know about any given (knot, rkey, workflow) so the +// order is moot. +func (r *providerRouter) Logs( + ctx context.Context, + knot string, + pipelineRkey string, + workflow string, +) (<-chan LogLine, error) { + for _, p := range r.providers { + ch, err := p.Logs(ctx, knot, pipelineRkey, workflow) + if errors.Is(err, ErrLogsNotFound) { + continue + } + return ch, err + } + return nil, ErrLogsNotFound +} + +// pick decodes raw and returns the first registered provider whose +// key appears as a child of the top-level `tack:` map, walking the +// YAML's children in document order. An empty body or YAML with no +// `tack:` block — or one whose children name no registered provider +// — is a structural error; no routing decision is possible. +func (r *providerRouter) pick(raw string) (Provider, error) { + if strings.TrimSpace(raw) == "" { + return nil, errors.New("workflow body is empty") + } + // MapSlice preserves the on-the-wire ordering of the children + // of `tack:` so "first match" is deterministic w.r.t. the YAML + // document, not Go's randomised map iteration. + var doc struct { + Tack yaml.MapSlice `yaml:"tack"` + } + if err := yaml.Unmarshal([]byte(raw), &doc); err != nil { + return nil, fmt.Errorf("parse workflow yaml: %w", err) + } + for _, item := range doc.Tack { + // YAML map keys are usually strings, but the lexicon + // doesn't enforce that — guard so a stray int/bool key + // doesn't panic the type assertion. + key, ok := item.Key.(string) + if !ok { + continue + } + if p, ok := r.providers[key]; ok { + return p, nil + } + } + return nil, fmt.Errorf( + "workflow yaml has no `tack:` key matching a registered provider", + ) +} diff --git a/provider_router_test.go b/provider_router_test.go new file mode 100644 --- /dev/null +++ b/provider_router_test.go @@ -0,0 +1,229 @@ +package main + +// Tests for providerRouter. We use a tiny in-test stub Provider — +// stubProvider — that records every Spawn call and serves canned +// Logs responses, so the tests stay focused on routing behaviour +// (which provider got called for which workflow YAML, and how +// ErrLogsNotFound is fanned out) without dragging in either the +// fake or Buildkite providers' end-to-end machinery. + +import ( + "context" + "errors" + "log/slog" + "sync" + "testing" + + "tangled.org/core/api/tangled" +) + +// stubProvider is a minimal Provider for routing tests. spawnCalls +// captures the workflow names handed to Spawn (single-element slices +// per the router's contract) so a test can assert which provider got +// which workflow. logsErr / logsCh govern what Logs returns; the +// default (zero value) is ErrLogsNotFound + nil channel, which makes +// fan-out tests easy to express by overriding only the provider that +// should "claim" the request. +type stubProvider struct { + mu sync.Mutex + spawnCalls []string + + logsErr error + logsCh chan LogLine +} + +var _ Provider = (*stubProvider)(nil) + +func (s *stubProvider) Spawn( + _ context.Context, + _ string, + _ string, + _ *tangled.Pipeline_TriggerMetadata, + workflows []*tangled.Pipeline_Workflow, +) { + s.mu.Lock() + defer s.mu.Unlock() + for _, wf := range workflows { + if wf == nil { + continue + } + s.spawnCalls = append(s.spawnCalls, wf.Name) + } +} + +func (s *stubProvider) Logs( + _ context.Context, + _ string, + _ string, + _ string, +) (<-chan LogLine, error) { + if s.logsErr != nil { + return nil, s.logsErr + } + if s.logsCh != nil { + return s.logsCh, nil + } + return nil, ErrLogsNotFound +} + +// names returns a defensive copy of spawnCalls so the test can read +// it without racing the router's per-workflow loop. The router calls +// Spawn synchronously in the test process, but a copy is the safer +// pattern if that ever changes. +func (s *stubProvider) names() []string { + s.mu.Lock() + defer s.mu.Unlock() + out := make([]string, len(s.spawnCalls)) + copy(out, s.spawnCalls) + return out +} + +// newRouterTest wires a router with a fixed pair of stubs ("a", "b") +// so tests can focus on the YAML → provider mapping. The stubs are +// returned alongside so each case can inspect what it received. +func newRouterTest() (*providerRouter, *stubProvider, *stubProvider) { + a := &stubProvider{} + b := &stubProvider{} + r := newProviderRouter(slog.Default(), map[string]Provider{ + "a": a, + "b": b, + }) + return r, a, b +} + +// TestProviderRouterSpawnRoutesByYAMLKey exercises the basic happy +// path: each workflow's `tack:` block names exactly one provider key, +// and the router hands that workflow to the matching provider only. +func TestProviderRouterSpawnRoutesByYAMLKey(t *testing.T) { + r, a, b := newRouterTest() + + r.Spawn(context.Background(), "knot", "rkey", nil, + []*tangled.Pipeline_Workflow{ + {Name: "wf-a.yml", Raw: "tack:\n a: {}\n"}, + {Name: "wf-b.yml", Raw: "tack:\n b: {}\n"}, + }, + ) + + if got, want := a.names(), []string{"wf-a.yml"}; !equalStrings(got, want) { + t.Fatalf("provider a got %v; want %v", got, want) + } + if got, want := b.names(), []string{"wf-b.yml"}; !equalStrings(got, want) { + t.Fatalf("provider b got %v; want %v", got, want) + } +} + +// TestProviderRouterSpawnFirstYAMLKeyWins pins tie-breaking when a +// workflow lists multiple provider keys: the YAML's document-order +// child of `tack:` wins, regardless of the Go map's iteration order. +func TestProviderRouterSpawnFirstYAMLKeyWins(t *testing.T) { + r, a, b := newRouterTest() + + // `b` is listed first under `tack:` so it should claim the + // workflow even though both keys are registered. + r.Spawn(context.Background(), "knot", "rkey", nil, + []*tangled.Pipeline_Workflow{ + {Name: "both.yml", Raw: "tack:\n b: {}\n a: {}\n"}, + }, + ) + + if got := a.names(); len(got) != 0 { + t.Fatalf("provider a should not have been called; got %v", got) + } + if got, want := b.names(), []string{"both.yml"}; !equalStrings(got, want) { + t.Fatalf("provider b got %v; want %v", got, want) + } +} + +// TestProviderRouterSpawnSkipsUnroutable confirms that workflows +// whose YAML has no matching provider key are skipped (logged but +// not dispatched) and don't poison the rest of the batch. +func TestProviderRouterSpawnSkipsUnroutable(t *testing.T) { + r, a, b := newRouterTest() + + r.Spawn(context.Background(), "knot", "rkey", nil, + []*tangled.Pipeline_Workflow{ + // No `tack:` key at all. + {Name: "bare.yml", Raw: "steps: []\n"}, + // `tack:` present but with an unknown sub-key. + {Name: "unknown.yml", Raw: "tack:\n nope: {}\n"}, + // Empty body — also unroutable. + {Name: "empty.yml", Raw: ""}, + // And one good one to prove the loop kept going. + {Name: "good.yml", Raw: "tack:\n a: {}\n"}, + }, + ) + + if got, want := a.names(), []string{"good.yml"}; !equalStrings(got, want) { + t.Fatalf("provider a got %v; want %v", got, want) + } + if got := b.names(); len(got) != 0 { + t.Fatalf("provider b should not have been called; got %v", got) + } +} + +// TestProviderRouterLogsFanOut verifies that Logs walks the +// providers and returns the channel from the first one that doesn't +// say ErrLogsNotFound. We seed exactly one provider with a real +// channel; map iteration order is unspecified but only one provider +// can possibly answer, so the test is deterministic. +func TestProviderRouterLogsFanOut(t *testing.T) { + r, _, b := newRouterTest() + + want := make(chan LogLine) + b.logsCh = want + + got, err := r.Logs(context.Background(), "k", "p", "w") + if err != nil { + t.Fatalf("Logs: %v", err) + } + if got != (<-chan LogLine)(want) { + t.Fatalf("got channel %v; want %v", got, want) + } +} + +// TestProviderRouterLogsAllNotFound makes sure that when no provider +// claims the tuple, the router surfaces ErrLogsNotFound itself — +// this is what the HTTP handler maps to a 404. +func TestProviderRouterLogsAllNotFound(t *testing.T) { + r, _, _ := newRouterTest() + + ch, err := r.Logs(context.Background(), "k", "p", "w") + if !errors.Is(err, ErrLogsNotFound) { + t.Fatalf("err = %v; want ErrLogsNotFound", err) + } + if ch != nil { + t.Fatalf("channel should be nil on not-found") + } +} + +// TestProviderRouterLogsBackendError confirms that a non-NotFound +// error from a provider is returned verbatim instead of being +// masked by the fan-out — backend failures must reach the HTTP +// caller as 5xx, not be silently retried elsewhere. +func TestProviderRouterLogsBackendError(t *testing.T) { + a := &stubProvider{logsErr: errors.New("boom")} + r := newProviderRouter(slog.Default(), map[string]Provider{"a": a}) + + ch, err := r.Logs(context.Background(), "k", "p", "w") + if err == nil || err.Error() != "boom" { + t.Fatalf("err = %v; want boom", err) + } + if ch != nil { + t.Fatalf("channel should be nil on backend error") + } +} + +// equalStrings is a small helper to compare ordered string slices — +// the router preserves workflow order within Spawn, so the tests +// assert against ordered slices rather than sets. +func equalStrings(a, b []string) bool { + if len(a) != len(b) { + return false + } + for i := range a { + if a[i] != b[i] { + return false + } + } + return true +} -- tangled.sh