amadan
console: pipeline API fixes — audit production side-write, classify SetPipeline errors
Review findings on the pipeline API (GET/PUT) commit: - ensureProductionForBakes now audits its own SetProduction flip, matching every other SetProduction call site in this package. Widened to take *Account so it can call s.audit. - SetPipeline errors are classified instead of echoed wholesale: ErrCASConflict -> 409 (re-read and retry), store/tombstone read failures -> 502 (never leaks bucket key paths), validatePipeline refusals -> 422 (the only case still safe to echo verbatim). - Pipeline audit row now names substance (channel names + count) instead of a bare "set the pipeline". New tests: production-side-write audit row + no-second-row on a following all-zero-bake PUT, pipeline audit detail names channels, and a 409/502 classification pair (stale-etag CAS conflict via a store wrapper, broken-store error-text non-leak).
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2 files changed,
+221
−12
internal/console/apipipeline.go+88 −12internal/console/apipipeline_test.go+133 −0
diff --git a/internal/console/apipipeline.go b/internal/console/apipipeline.go| index 3cb1de236..2079084bd 100644 |
| --- a/internal/console/apipipeline.go |
| +++ b/internal/console/apipipeline.go |
| @@ -26,10 +26,12 @@ package console |
| import ( |
| "context" |
| + "errors" |
| "fmt" |
| "log" |
| "net/http" |
| "net/url" |
| + "strings" |
| "time" |
| "github.com/carlosframework/platform/internal/release" |
| @@ -92,8 +94,12 @@ func (s *Server) handleAPIPipelineGet(w http.ResponseWriter, r *http.Request) { |
| // quorum flow. |
| // SetPipeline — validates shape internally (canary reserved, |
| // duplicate names, bad bake strings, negative |
| -// approvals) and signs; its error becomes 422, |
| -// the caller's to fix. |
| +// approvals) and signs. Its error is CLASSIFIED |
| +// (classifySetPipelineError, below) rather than |
| +// echoed wholesale: a validation refusal is 422, |
| +// the caller's to fix; a lost CAS race is 409, |
| +// re-ask; a store read/write failure is 502 and |
| +// never puts a bucket key path on the wire. |
| // production side-write — spec §3's old-binary fallback, below. |
| // audit + bump — after the write lands, never before or in |
| // place of a gate above. |
| @@ -148,11 +154,20 @@ func (s *Server) handleAPIPipelinePut(w http.ResponseWriter, r *http.Request) { |
| } |
| set, err := release.SetPipeline(ctx, s.cfg.Releases, s.cfg.Key, acct.Sqid, app, in.Channels, addr) |
| if err != nil { |
| - jsonError(w, http.StatusUnprocessableEntity, err.Error()) |
| + status, msg := classifySetPipelineError(err) |
| + if status != http.StatusUnprocessableEntity { |
| + // A validation refusal is the caller's own typo — the message |
| + // already says so, on the wire, and logging every bad request a |
| + // script sends would just be noise. A 409 or 502 here is a |
| + // genuine backend condition (a race, an unreachable store) an |
| + // operator may need the real detail for. |
| + log.Printf("console: api pipeline put %s/%s: set: %v", acct.Sqid, app, err) |
| + } |
| + jsonError(w, status, msg) |
| return |
| } |
| - s.ensureProductionForBakes(ctx, acct.Sqid, app, set, addr) |
| - s.audit(acct, app, addr, "pipeline", "set the pipeline (cli)") |
| + s.ensureProductionForBakes(ctx, acct, app, set, addr) |
| + s.audit(acct, app, addr, "pipeline", pipelineAuditDetail(set)) |
| // Box-consumed (carlos-deploy §3): boxes read pipeline.json on the same |
| // fast pass channel pointers and config already ride. |
| s.bumpSerial(ctx) |
| @@ -206,10 +221,15 @@ func guardedEdit(cur *release.Pipeline, next []release.PipelineChannel) (string, |
| // re-declared: GetProduction's error already fails every reader toward |
| // "treat as production" (adopt path, trash's delete gate), so writing over |
| // it here would only risk losing whatever the existing object actually |
| -// says. Log-only on every failure: the pipeline write itself already |
| +// says. Log-only on a READ failure: the pipeline write itself already |
| // landed, and a box that never sees this flag flip is no worse off than a |
| -// pipeline-less app was yesterday. |
| -func (s *Server) ensureProductionForBakes(ctx context.Context, acct, app string, p *release.Pipeline, actor string) { |
| +// pipeline-less app was yesterday. A successful FLIP, though, gets the same |
| +// audit row every other SetProduction call site writes (server.go's |
| +// enable_bake and production-toggle doors) — this flag gates fleet-wide |
| +// baking and delete protection, so a row that goes unaudited here while |
| +// every other path to it is audited would be the one silent way to turn it |
| +// on. |
| +func (s *Server) ensureProductionForBakes(ctx context.Context, acct *Account, app string, p *release.Pipeline, actor string) { |
| bakes := false |
| for _, c := range p.Channels { |
| if d, err := c.BakeWindow(); err == nil && d > 0 { |
| @@ -220,17 +240,73 @@ func (s *Server) ensureProductionForBakes(ctx context.Context, acct, app string, |
| if !bakes { |
| return |
| } |
| - flag, err := release.GetProduction(ctx, s.cfg.Releases, s.cfg.ReleasePub, acct, app) |
| + flag, err := release.GetProduction(ctx, s.cfg.Releases, s.cfg.ReleasePub, acct.Sqid, app) |
| if err != nil { |
| - log.Printf("console: pipeline %s/%s: read production flag for bake fallback: %v", acct, app, err) |
| + log.Printf("console: pipeline %s/%s: read production flag for bake fallback: %v", acct.Sqid, app, err) |
| return |
| } |
| if flag != nil && flag.Production { |
| return // already true — never re-declared, never downgraded |
| } |
| - if err := release.SetProduction(ctx, s.cfg.Releases, s.cfg.Key, acct, app, true, actor); err != nil { |
| - log.Printf("console: pipeline %s/%s: set production for bake fallback: %v", acct, app, err) |
| + if err := release.SetProduction(ctx, s.cfg.Releases, s.cfg.Key, acct.Sqid, app, true, actor); err != nil { |
| + log.Printf("console: pipeline %s/%s: set production for bake fallback: %v", acct.Sqid, app, err) |
| + return |
| + } |
| + s.audit(acct, app, actor, "production", |
| + "enabled the safety delay (production: on, at pipeline set — old-binary bake fallback)") |
| +} |
| + |
| +// classifySetPipelineError maps a release.SetPipeline error to the status |
| +// and caller-facing message this door answers with. SetPipeline's errors |
| +// come in three shapes, and only one of them is safe to echo verbatim: |
| +// |
| +// - release.ErrCASConflict (checked by errors.Is, since SetPipeline |
| +// returns it — via Store.Put — unwrapped): a concurrent write won the |
| +// race. 409, with a fixed retry sentence — not the caller's fault and |
| +// not a defect in what they sent. |
| +// - A store or trash-marker READ failure, wrapped with the exact prefixes |
| +// SetPipeline's own source uses ("cannot check the trash marker" / |
| +// "set pipeline: read "): the bucket, not the request, is the problem. |
| +// 502, with a fixed sentence — the real error (bucket names, request |
| +// ids) stays in the log, the same discipline every other store-backed |
| +// door here follows (e.g. handleAPITrashApp, apiledger.go). |
| +// - Everything else prefixed "set pipeline: " is validatePipeline's own |
| +// refusal (or one of the pre-checks that cannot fire from this door, |
| +// since apiAccountApp already validated the account/app names) — |
| +// genuinely the caller's malformed input, safe and useful to echo. 422. |
| +// - An error with NO "set pipeline: " prefix at all and not ErrCASConflict |
| +// is the one other shape SetPipeline returns as-is: a raw Store.Put |
| +// failure. Not the caller's fault either — 502, generic. |
| +func classifySetPipelineError(err error) (int, string) { |
| + if errors.Is(err, release.ErrCASConflict) { |
| + return http.StatusConflict, "concurrent change — re-read and retry" |
| + } |
| + msg := err.Error() |
| + if strings.Contains(msg, "cannot check the trash marker") || strings.Contains(msg, "set pipeline: read ") { |
| + return http.StatusBadGateway, "could not verify the current pipeline — try again" |
| + } |
| + if strings.HasPrefix(msg, "set pipeline: ") { |
| + return http.StatusUnprocessableEntity, msg |
| + } |
| + return http.StatusBadGateway, "could not write the pipeline — try again" |
| +} |
| + |
| +// pipelineAuditDetail names what actually changed, in the house style the |
| +// neighbouring audit rows use (server.go's promote/production rows): a |
| +// bare "set the pipeline" would leave a reader of the audit trail unable to |
| +// tell a one-line typo fix from a whole new ladder without opening the |
| +// bucket object themselves — and this row documents a POLICY rewrite, not a |
| +// routine write like a config var. |
| +func pipelineAuditDetail(p *release.Pipeline) string { |
| + names := make([]string, len(p.Channels)) |
| + for i, c := range p.Channels { |
| + names[i] = c.Name |
| + } |
| + suffix := "s" |
| + if len(p.Channels) == 1 { |
| + suffix = "" |
| } |
| + return fmt.Sprintf("set the pipeline (%s; %d channel%s, cli)", strings.Join(names, " → "), len(p.Channels), suffix) |
| } |
| // holdWindow is the pipeline-aware form of release.HoldFor (bake.go), for |
diff --git a/internal/console/apipipeline_test.go b/internal/console/apipipeline_test.go| index be23f50df..186c4be4c 100644 |
| --- a/internal/console/apipipeline_test.go |
| +++ b/internal/console/apipipeline_test.go |
| @@ -229,6 +229,139 @@ func TestAPIPipelinePutGuardedChannelInterimRefusal(t *testing.T) { |
| } |
| } |
| +// TestAPIPipelinePutAuditsSubstanceAndProductionSideWrite: two audit |
| +// findings from review. First, the pipeline write's own row must name |
| +// substance (channel names and count), not a bare "set the pipeline" that |
| +// forces a reader to open the bucket object to see what changed. Second, |
| +// the production-flag side-write (spec §3's old-binary fallback) is a real |
| +// SetProduction call — every OTHER call site in this codebase |
| +// (server.go's enable_bake and production-toggle doors) writes its own |
| +// audit row when it flips the flag, and this one must too: the flag gates |
| +// fleet-wide baking and delete protection, so a silent flip here would be |
| +// the one path to it nothing in the audit trail explains. |
| +func TestAPIPipelinePutAuditsSubstanceAndProductionSideWrite(t *testing.T) { |
| + fx := newConsoleFixture(t) |
| + token := cliToken(t, fx, "paul@keymail.dev", "code-pipeline-audit") |
| + if _, err := fx.st.CreateAppIn(OpsAccountID, "widget", "paul@keymail.dev", time.Now()); err != nil { |
| + t.Fatal(err) |
| + } |
| + |
| + status, out := putJSON(t, fx.ts.URL+"/api/cli/apps/bab/widget/pipeline", |
| + map[string]any{"channels": []map[string]any{ |
| + {"name": "edge"}, |
| + {"name": "production", "bake": "24h"}, |
| + }}, token) |
| + if status != http.StatusOK { |
| + t.Fatalf("pipeline put = %d %v", status, out) |
| + } |
| + |
| + events, err := fx.st.Events(OpsAccountID, "widget", 10) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + var pipelineRow, productionRow *Event |
| + for i := range events { |
| + switch events[i].Action { |
| + case "pipeline": |
| + pipelineRow = &events[i] |
| + case "production": |
| + productionRow = &events[i] |
| + } |
| + } |
| + if pipelineRow == nil { |
| + t.Fatalf("events = %v, want a pipeline audit row", events) |
| + } |
| + if !strings.Contains(pipelineRow.Detail, "edge") || !strings.Contains(pipelineRow.Detail, "production") || |
| + !strings.Contains(pipelineRow.Detail, "2 channel") { |
| + t.Fatalf("pipeline audit detail = %q, want it to name the channels and count", pipelineRow.Detail) |
| + } |
| + if productionRow == nil { |
| + t.Fatalf("events = %v, want a production audit row for the side-write", events) |
| + } |
| + if !strings.Contains(productionRow.Detail, "on") { |
| + t.Fatalf("production audit detail = %q, want it to say the flag went on", productionRow.Detail) |
| + } |
| + |
| + // A second PUT that declares no bake at all must not flip production |
| + // again — and therefore must not write a second production audit row. |
| + status, out = putJSON(t, fx.ts.URL+"/api/cli/apps/bab/widget/pipeline", |
| + map[string]any{"channels": []map[string]any{{"name": "edge"}}}, token) |
| + if status != http.StatusOK { |
| + t.Fatalf("all-zero-bake pipeline put = %d %v", status, out) |
| + } |
| + events, err = fx.st.Events(OpsAccountID, "widget", 10) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + productionRows := 0 |
| + for _, e := range events { |
| + if e.Action == "production" { |
| + productionRows++ |
| + } |
| + } |
| + if productionRows != 1 { |
| + t.Fatalf("production audit rows after an all-zero-bake put = %d, want still 1 (never re-declared)", productionRows) |
| + } |
| +} |
| + |
| +// pipelineCASStore forces SetPipeline's own conditional Put on pipeline.json |
| +// to lose the race — the same shape casStore (apiledger_test.go) uses for |
| +// ledger keys — simulating a concurrent pipeline write without needing two |
| +// real goroutines to land in the narrow window between SetPipeline's own |
| +// read and its write. |
| +type pipelineCASStore struct{ release.Store } |
| + |
| +func (c pipelineCASStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) { |
| + if strings.HasSuffix(key, "pipeline.json") { |
| + return "", release.ErrCASConflict |
| + } |
| + return c.Store.Put(ctx, key, body, ifMatch) |
| +} |
| + |
| +// TestAPIPipelinePutStoreFailureStatuses: SetPipeline's non-validation |
| +// failures must not be echoed wholesale — a lost CAS race is retry-worthy |
| +// (409, a fixed sentence), and a broken store is not the caller's fault and |
| +// must never put the store's own error text (bucket names, request ids) on |
| +// the wire (502, generic). |
| +func TestAPIPipelinePutStoreFailureStatuses(t *testing.T) { |
| + t.Run("concurrent write is 409", func(t *testing.T) { |
| + fx := newConsoleFixture(t) |
| + if _, err := fx.st.CreateAppIn(OpsAccountID, "widget", "paul@keymail.dev", time.Now()); err != nil { |
| + t.Fatal(err) |
| + } |
| + token := cliToken(t, fx, "paul@keymail.dev", "code-pipeline-cas") |
| + fx.srv.cfg.Releases = pipelineCASStore{fx.rel} |
| + |
| + status, out := putJSON(t, fx.ts.URL+"/api/cli/apps/bab/widget/pipeline", |
| + map[string]any{"channels": []map[string]any{{"name": "edge"}}}, token) |
| + if status != http.StatusConflict { |
| + t.Fatalf("pipeline put racing a concurrent write = %d %v, want 409", status, out) |
| + } |
| + if !strings.Contains(str(t, out, "error"), "re-read and retry") { |
| + t.Fatalf("cas conflict error = %v, want it to say re-read and retry", out["error"]) |
| + } |
| + }) |
| + |
| + t.Run("broken store is 502 and never leaks its error text", func(t *testing.T) { |
| + fx := newConsoleFixture(t) |
| + if _, err := fx.st.CreateAppIn(OpsAccountID, "widget", "paul@keymail.dev", time.Now()); err != nil { |
| + t.Fatal(err) |
| + } |
| + token := cliToken(t, fx, "paul@keymail.dev", "code-pipeline-broken") |
| + fx.srv.cfg.Releases = brokenStore{fx.rel} |
| + |
| + status, out := putJSON(t, fx.ts.URL+"/api/cli/apps/bab/widget/pipeline", |
| + map[string]any{"channels": []map[string]any{{"name": "edge"}}}, token) |
| + if status != http.StatusBadGateway { |
| + t.Fatalf("pipeline put against a broken store = %d %v, want 502", status, out) |
| + } |
| + if strings.Contains(str(t, out, "error"), "AccessDenied") || |
| + strings.Contains(str(t, out, "error"), "carlos-prod-secret-bucket") { |
| + t.Fatalf("broken-store error leaked the store's own text: %v", out["error"]) |
| + } |
| + }) |
| +} |
| + |
| // TestAPIPipelinePutNoServiceRoleReachesIt: svcrole.Allow() with NO roles is |
| // the fully-closed Set — Admin included (svcrole.TestEmptySetPermitsNothingIncludingAdmin) |
| // — so the door must be registered svcrole.Allow(svcrole.Admin), not |