diff --git a/e2e/scenarios/hotfix/hotfix-generation-threshold.yaml b/e2e/scenarios/hotfix/hotfix-generation-threshold.yaml index 674b77e0..8f516805 100644 --- a/e2e/scenarios/hotfix/hotfix-generation-threshold.yaml +++ b/e2e/scenarios/hotfix/hotfix-generation-threshold.yaml @@ -40,7 +40,8 @@ steps: - "pull_request:" - "types: [closed]" - "'env/*'" - - "group: hotfix-" + - "format('hotfix-finalize-{0}', github.repository)" + - "format('hotfix-{0}', github.event.inputs.target_env)" - " plan:" - " apply:" - " check:" diff --git a/internal/generate/hotfix.go b/internal/generate/hotfix.go index e49fefba..c1bde2c7 100644 --- a/internal/generate/hotfix.go +++ b/internal/generate/hotfix.go @@ -183,11 +183,21 @@ func (g *HotfixGenerator) writePermissions(sb *strings.Builder) { writeTopLevelPermissions(sb, base) } -// writeConcurrency keys the group per target environment. On dispatch the env is -// the operator input; on pull_request close it is derived from the base ref. +// writeConcurrency keys the apply (dispatch) path per target environment, but +// keys the finalize (pull_request close) path on a per-repository constant so +// concurrent per-environment finalize runs QUEUE rather than race. +// +// Each env's finalize commits the manifest to trunk through the Contents API. +// Keyed per base ref, two envs whose resolution PRs close together fall into +// different concurrency groups and PUT in parallel; the second writer's blob SHA +// is stale and GitHub returns 409, dropping that env's state. A per-repo finalize +// group with cancel-in-progress: false serializes those writes instead. This is +// defense-in-depth: the durable fix is the Contents API 409 read-modify-write +// retry in internal/statewrite, which still protects against any other writer +// (orchestrate, promote, rollback) that a manifest-global group cannot serialize. func (g *HotfixGenerator) writeConcurrency(sb *strings.Builder) { sb.WriteString("concurrency:\n") - sb.WriteString(" group: hotfix-${{ github.event.inputs.target_env || github.event.pull_request.base.ref }}\n") + sb.WriteString(" group: ${{ github.event_name == 'pull_request' && format('hotfix-finalize-{0}', github.repository) || format('hotfix-{0}', github.event.inputs.target_env) }}\n") sb.WriteString(" cancel-in-progress: false\n") sb.WriteString("\n") } diff --git a/internal/generate/hotfix_test.go b/internal/generate/hotfix_test.go index b285b9e3..16b5eaae 100644 --- a/internal/generate/hotfix_test.go +++ b/internal/generate/hotfix_test.go @@ -172,7 +172,16 @@ func TestHotfixGenerator_Concurrency(t *testing.T) { gen := NewHotfixGenerator(threeEnvHotfixConfig(), "") content, err := gen.Generate() require.NoError(t, err) - assert.Contains(t, content, "group: hotfix-") + + // The finalize (pull_request close) path keys on a per-repository constant so + // concurrent per-environment finalize runs queue instead of racing on the + // shared manifest blob SHA. The apply (dispatch) path stays keyed per target + // environment so unrelated cherry-picks still run in parallel. + assert.Contains(t, content, + "github.event_name == 'pull_request' && format('hotfix-finalize-{0}', github.repository)", + "finalize must use a per-repository concurrency group so writes queue") + assert.Contains(t, content, "format('hotfix-{0}', github.event.inputs.target_env)", + "the dispatch path must stay keyed per target environment") assert.Contains(t, content, "cancel-in-progress: false") } diff --git a/internal/hotfix/finalize.go b/internal/hotfix/finalize.go index 8740e8c7..b2133cf9 100644 --- a/internal/hotfix/finalize.go +++ b/internal/hotfix/finalize.go @@ -1,7 +1,6 @@ package hotfix import ( - "encoding/base64" "fmt" "os" "os/exec" @@ -13,6 +12,7 @@ import ( "github.com/stablekernel/cascade/internal/config" "github.com/stablekernel/cascade/internal/git" "github.com/stablekernel/cascade/internal/release" + "github.com/stablekernel/cascade/internal/statewrite" "github.com/stablekernel/cascade/internal/version" ) @@ -135,51 +135,39 @@ func (execTagLister) ListTags() ([]string, error) { type gitStatePusher struct{} func (gitStatePusher) CommitAndPush(path, branch, message string) error { - if isRealGitHub() { - return writeStateViaAPI(path, branch, message) - } return commitAndPushGit(path, branch, message) } -// isRealGitHub reports whether the workflow runs on github.com rather than an -// act/gitea e2e environment, detected by GITHUB_SERVER_URL as the generated -// dispatch steps do. -func isRealGitHub() bool { - server := os.Getenv("GITHUB_SERVER_URL") - return server == "" || server == "https://github.com" +// apiStatePusher commits the manifest to trunk through the GitHub Contents REST +// API using the shared optimistic-lock retry loop, so concurrent env finalizers +// that each touch only their own env state merge rather than clobbering each +// other on the file blob SHA. mutate re-applies this hotfix's state change onto +// whatever trunk bytes the loop fetches. +type apiStatePusher struct { + mutate statewrite.Mutate } -// writeStateViaAPI writes the manifest to the trunk branch through the GitHub -// Contents REST API using the gh CLI, producing a signed (Verified) commit. -func writeStateViaAPI(path, branch, message string) error { +func (p apiStatePusher) CommitAndPush(path, branch, message string) error { repo := os.Getenv("GITHUB_REPOSITORY") if repo == "" { return fmt.Errorf("GITHUB_REPOSITORY is not set; cannot write state via API") } + return statewrite.CommitWithRetry(statewrite.Options{ + Client: statewrite.NewGHClient(), + Repo: repo, + Path: path, + Ref: branch, + Message: message, + Mutate: p.mutate, + }) +} - data, err := os.ReadFile(path) - if err != nil { - return fmt.Errorf("read manifest failed: %w", err) - } - contentB64 := base64.StdEncoding.EncodeToString(data) - apiPath := fmt.Sprintf("repos/%s/contents/%s", repo, path) - - shaOut, _ := exec.Command("gh", "api", fmt.Sprintf("%s?ref=%s", apiPath, branch), "--jq", ".sha").Output() - currentSHA := strings.TrimSpace(string(shaOut)) - - args := []string{ - "api", apiPath, "-X", "PUT", - "-f", "message=" + message, - "-f", "content=" + contentB64, - "-f", "branch=" + branch, - } - if currentSHA != "" { - args = append(args, "-f", "sha="+currentSHA) - } - if out, err := exec.Command("gh", args...).CombinedOutput(); err != nil { - return fmt.Errorf("state write via API failed: %s: %w", strings.TrimSpace(string(out)), err) - } - return nil +// isRealGitHub reports whether the workflow runs on github.com rather than an +// act/gitea e2e environment, detected by GITHUB_SERVER_URL as the generated +// dispatch steps do. +func isRealGitHub() bool { + server := os.Getenv("GITHUB_SERVER_URL") + return server == "" || server == "https://github.com" } // commitAndPushGit commits the manifest and pushes it to the trunk branch with @@ -227,6 +215,11 @@ type Finalizer struct { pusher statePusher tipReader gitTipReader trunkReader trunkStateReader + + // pusherInjected records whether a caller supplied an explicit statePusher. + // When true, Finalize uses that pusher verbatim (tests inject a recorder); + // when false on real GitHub, Finalize swaps in the API retry pusher. + pusherInjected bool } // FinalizerOptions carries the required inputs for NewFinalizer. @@ -270,6 +263,7 @@ func WithStatePusher(p statePusher) FinalizeOption { return func(f *Finalizer) { if p != nil { f.pusher = p + f.pusherInjected = true } } } @@ -433,32 +427,45 @@ func (f *Finalizer) Finalize(targetEnv, mergeSHA string, fixSHAs []string, baseS return nil } - // Snapshot the prior state into the deploy-history ring (newest first, - // bounded). The idempotency gate above already returned when the state - // records mergeSHA, so this records a genuine transition; the gate inside - // PushPreviousSnapshot is belt-and-suspenders. - prior.PushPreviousSnapshot(mergeSHA) - - // Carry BaseSHA forward when already diverged; otherwise anchor it now. - if prior.BaseSHA == "" { - prior.BaseSHA = baseSHA + // Apply the state mutation in place onto the trunk manifest, snapshotting the + // prior state into the Previous ring and writing the divergence fields and + // substates. Extracted so the same change can be re-applied against freshly + // fetched trunk bytes inside the optimistic-lock retry loop below. + if err := f.applyHotfixState(f.cicd, targetEnv, mergeSHA, hotfixVersion, baseSHA, timestamp, fixSHAs); err != nil { + return err } - prior.Patches = append(prior.Patches, fixSHAs...) - prior.Ref = branch - prior.SHA = mergeSHA - prior.Version = hotfixVersion - prior.CommittedAt = timestamp - prior.CommittedBy = f.actor - - // Record per-deploy and per-build substates for successful jobs. - f.recordSubstates(prior, mergeSHA, hotfixVersion, timestamp) if err := f.writeConfig(); err != nil { return err } message := fmt.Sprintf("chore: record hotfix %s on %s [skip ci]", hotfixVersion, targetEnv) - if err := f.pusher.CommitAndPush(f.configPath, trunk, message); err != nil { + + pusher := f.pusher + if !f.pusherInjected && isRealGitHub() { + capturedVersion := hotfixVersion + capturedTimestamp := timestamp + capturedBaseSHA := baseSHA + capturedFixSHAs := append([]string(nil), fixSHAs...) + capturedTarget := targetEnv + capturedMerge := mergeSHA + key := f.manifestKey + pusher = apiStatePusher{mutate: func(current []byte) ([]byte, error) { + fresh, err := config.ParseManifestBytes(current, key) + if err != nil { + return nil, fmt.Errorf("parsing current manifest: %w", err) + } + if err := f.applyHotfixState(fresh, capturedTarget, capturedMerge, capturedVersion, capturedBaseSHA, capturedTimestamp, capturedFixSHAs); err != nil { + return nil, err + } + data, err := yaml.Marshal(map[string]any{key: fresh}) + if err != nil { + return nil, fmt.Errorf("marshaling merged manifest: %w", err) + } + return data, nil + }} + } + if err := pusher.CommitAndPush(f.configPath, trunk, message); err != nil { return fmt.Errorf("committing hotfix state: %w", err) } @@ -471,6 +478,37 @@ func (f *Finalizer) Finalize(targetEnv, mergeSHA string, fixSHAs []string, baseS return nil } +// applyHotfixState applies the hotfix state mutation for targetEnv onto cicd +// using pre-computed values, so it can be re-applied against freshly fetched +// trunk bytes inside the optimistic-lock retry loop. It is idempotent: when the +// manifest already records mergeSHA for targetEnv the call is a no-op, so a retry +// (or rerun) neither double-appends patches nor re-snapshots the Previous ring. +func (f *Finalizer) applyHotfixState(cicd *config.CICDFile, targetEnv, mergeSHA, hotfixVersion, baseSHA, timestamp string, fixSHAs []string) error { + if cicd.State == nil { + cicd.State = make(map[string]*config.EnvState) + } + prior := cicd.State[targetEnv] + if prior == nil { + prior = &config.EnvState{} + cicd.State[targetEnv] = prior + } + if prior.SHA == mergeSHA { + return nil + } + prior.PushPreviousSnapshot(mergeSHA) + if prior.BaseSHA == "" { + prior.BaseSHA = baseSHA + } + prior.Patches = append(prior.Patches, fixSHAs...) + prior.Ref = envBranch(targetEnv) + prior.SHA = mergeSHA + prior.Version = hotfixVersion + prior.CommittedAt = timestamp + prior.CommittedBy = f.actor + f.recordSubstates(prior, mergeSHA, hotfixVersion, timestamp) + return nil +} + // readTrunkManifest fetches the manifest as it exists on the trunk branch and // returns the parsed manifest. It is read from trunk because promote finalize // writes env state only to trunk; the env branch the hotfix merged into lags diff --git a/internal/hotfix/finalize_test.go b/internal/hotfix/finalize_test.go index 58da2c52..28eb7151 100644 --- a/internal/hotfix/finalize_test.go +++ b/internal/hotfix/finalize_test.go @@ -138,6 +138,33 @@ func loadState(t *testing.T, manifest, env string) *config.EnvState { return cicd.State[env] } +// TestFinalizeHotfixMutatePreservesUntouchedEnv verifies that applying the +// hotfix state mutation for one env on a manifest leaves every other env's +// recorded state untouched, so re-applying it against re-fetched trunk bytes in +// the optimistic-lock loop merges rather than clobbers concurrent writers. +func TestFinalizeHotfixMutatePreservesUntouchedEnv(t *testing.T) { + cicd := &config.CICDFile{State: map[string]*config.EnvState{ + "dev": {SHA: "dev-sha", Version: "v1.0.0"}, + "staging": {SHA: "stg-sha", Version: "v1.1.0"}, + }} + f := &Finalizer{actor: "tester", manifestKey: "ci"} + + err := f.applyHotfixState(cicd, "staging", "merge-sha", "v1.1.1", "base-sha", "2026-01-01T00:00:00Z", []string{"fix-sha"}) + if err != nil { + t.Fatalf("applyHotfixState: %v", err) + } + + if cicd.State["dev"].SHA != "dev-sha" { + t.Errorf("dev.sha = %q, want unchanged dev-sha", cicd.State["dev"].SHA) + } + if cicd.State["staging"].SHA != "merge-sha" { + t.Errorf("staging.sha = %q, want merge-sha", cicd.State["staging"].SHA) + } + if cicd.State["staging"].Version != "v1.1.1" { + t.Errorf("staging.version = %q, want v1.1.1", cicd.State["staging"].Version) + } +} + func TestFinalize_WritesDivergedState(t *testing.T) { newScratchRepo(t) base := commitFile(t, "a.txt", "one", "first") diff --git a/internal/promote/finalize.go b/internal/promote/finalize.go index f4d44742..e1733da7 100644 --- a/internal/promote/finalize.go +++ b/internal/promote/finalize.go @@ -1,7 +1,6 @@ package promote import ( - "encoding/base64" "fmt" "os" "os/exec" @@ -9,6 +8,7 @@ import ( "time" "github.com/stablekernel/cascade/internal/config" + "github.com/stablekernel/cascade/internal/statewrite" "gopkg.in/yaml.v3" ) @@ -365,44 +365,63 @@ func isRealGitHub() bool { } // writeStateViaAPI writes the manifest file to the trunk branch through the -// GitHub Contents REST API using the gh CLI. This produces a signed (Verified) -// commit and, with a bypass-capable token, can update a protected branch. +// GitHub Contents REST API using the shared optimistic-lock retry loop. This +// produces a signed (Verified) commit and, with a bypass-capable token, can +// update a protected branch. The mutation re-parses whatever trunk bytes the +// loop fetches and overlays only this finalizer's owned env state, so two +// concurrent env finalizers merge rather than clobber each other on the file +// blob SHA. func (f *Finalizer) writeStateViaAPI(message string) error { repo := os.Getenv("GITHUB_REPOSITORY") if repo == "" { return fmt.Errorf("GITHUB_REPOSITORY is not set; cannot write state via API") } branch := trunkBranchFromEnv() - - data, err := os.ReadFile(f.configPath) - if err != nil { - return fmt.Errorf("read manifest failed: %w", err) + key := config.DefaultManifestKey + if f.cicdFile.Config != nil && f.cicdFile.Config.ManifestKey != "" { + key = f.cicdFile.Config.ManifestKey } - contentB64 := base64.StdEncoding.EncodeToString(data) - - apiPath := fmt.Sprintf("repos/%s/contents/%s", repo, f.configPath) - - // Fetch the current blob SHA so the API performs an update rather than a - // create. An empty result means the file does not yet exist on the branch. - shaCmd := exec.Command("gh", "api", fmt.Sprintf("%s?ref=%s", apiPath, branch), "--jq", ".sha") - shaOut, _ := shaCmd.Output() - currentSHA := strings.TrimSpace(string(shaOut)) + return statewrite.CommitWithRetry(statewrite.Options{ + Client: statewrite.NewGHClient(), + Repo: repo, + Path: f.configPath, + Ref: branch, + Message: message, + Mutate: func(current []byte) ([]byte, error) { + into, err := config.ParseManifestBytes(current, key) + if err != nil { + return nil, fmt.Errorf("parsing current manifest: %w", err) + } + f.overlayOwnedState(into) + data, err := yaml.Marshal(map[string]any{key: into}) + if err != nil { + return nil, fmt.Errorf("marshaling merged manifest: %w", err) + } + return data, nil + }, + }) +} - args := []string{ - "api", apiPath, "-X", "PUT", - "-f", "message=" + message, - "-f", "content=" + contentB64, - "-f", "branch=" + branch, +// overlayOwnedState copies the state this finalizer owns from its in-memory, +// already-mutated manifest onto into, the freshly fetched trunk manifest. It +// overlays only the promoted envs (and, on a publish, the release marker and +// latest_release) so a concurrent finalizer's keys on into are preserved. It is +// re-appliable: CommitWithRetry calls it again against re-fetched trunk bytes on +// a 409, and each call deterministically re-derives the same owned keys. +func (f *Finalizer) overlayOwnedState(into *config.CICDFile) { + if into.State == nil { + into.State = make(map[string]*config.EnvState) } - if currentSHA != "" { - args = append(args, "-f", "sha="+currentSHA) + if f.promotionResult != nil { + for _, promo := range f.promotionResult.Promotions { + into.State[promo.Environment] = f.cicdFile.State[promo.Environment] + } } - - cmd := exec.Command("gh", args...) - if out, err := cmd.CombinedOutput(); err != nil { - return fmt.Errorf("state write via API failed: %s: %w", strings.TrimSpace(string(out)), err) + if f.promotionResult != nil && f.promotionResult.ReleaseAction == "publish" { + into.LatestRelease = f.cicdFile.LatestRelease + into.State["release"] = f.cicdFile.State["release"] + delete(into.State, "prerelease") } - return nil } // commitAndPushGit commits the manifest and pushes with plain git. Used in the diff --git a/internal/promote/finalize_test.go b/internal/promote/finalize_test.go index f56eb390..2e5d82ad 100644 --- a/internal/promote/finalize_test.go +++ b/internal/promote/finalize_test.go @@ -753,6 +753,35 @@ func TestCommitAndPushGit_DetachedHeadPushesToTrunk(t *testing.T) { require.Contains(t, string(out), "update state after promotion to test") } +// TestFinalizeOverlayMerge verifies that overlaying this finalizer's owned env +// state onto a freshly fetched trunk manifest preserves a concurrent writer's +// untouched env keys while applying this finalizer's promoted env. +func TestFinalizeOverlayMerge(t *testing.T) { + f := &Finalizer{ + cicdFile: &config.CICDFile{ + State: map[string]*config.EnvState{ + "staging": {SHA: "stg-sha", Version: "v2.0.0"}, + }, + }, + promotionResult: &PromotionResult{ + Promotions: []EnvPromotion{{Environment: "staging"}}, + }, + } + + current := &config.CICDFile{ + State: map[string]*config.EnvState{ + "dev": {SHA: "dev-sha"}, + }, + } + + f.overlayOwnedState(current) + + require.NotNil(t, current.State["dev"], "concurrent writer's env must survive") + require.Equal(t, "dev-sha", current.State["dev"].SHA, "untouched env must be preserved") + require.NotNil(t, current.State["staging"], "promoted env must be overlaid") + require.Equal(t, "stg-sha", current.State["staging"].SHA) +} + func TestUpdateState_PushesPriorSnapshotOnTransition(t *testing.T) { tmpDir := t.TempDir() configPath := filepath.Join(tmpDir, "manifest.yaml") diff --git a/internal/statewrite/apiwrite.go b/internal/statewrite/apiwrite.go new file mode 100644 index 00000000..1d1bdd78 --- /dev/null +++ b/internal/statewrite/apiwrite.go @@ -0,0 +1,170 @@ +// Package statewrite provides the shared optimistic-lock retry used by every +// finalize verb (orchestrate, promote, rollback, hotfix) when it commits the +// manifest to the trunk branch through the GitHub Contents REST API. +// +// Concurrent finalize jobs for different environments mutate disjoint keys in +// the same manifest file. They do not conflict semantically, but they collide +// on the file blob SHA: the second writer to PUT with a now-stale SHA gets an +// HTTP 409 ("does not match ") and its state is dropped. CommitWithRetry +// closes that race by expressing the write as a read-modify-write that re-reads +// the current manifest, re-applies the caller's mutation on top of whatever the +// other writer committed, and re-PUTs, retrying on 409. +package statewrite + +import ( + "fmt" + "strings" + "time" +) + +// maxAttempts bounds the read-modify-write retry loop. Five attempts comfortably +// absorbs the handful of envs that can finalize in parallel without masking a +// genuinely stuck write. +const maxAttempts = 5 + +// retryBackoff is the base delay between optimistic-lock retries. Attempt N +// waits N*retryBackoff so concurrent writers stagger rather than re-collide. +const retryBackoff = 500 * time.Millisecond + +// ContentsClient is the minimal GitHub Contents API surface the retry loop +// needs. The production implementation shells out to the gh CLI; tests inject a +// fake that returns a 409 on the first PUT and succeeds on the second. +// +// GetContent returns the current file bytes and its blob SHA at ref. A file that +// does not yet exist returns an empty SHA and a nil error so the first write +// creates it. PutContent writes content at ref using sha for the optimistic +// lock (empty sha creates the file); it returns an error satisfying IsConflict +// when the blob SHA no longer matches. +type ContentsClient interface { + GetContent(repo, path, ref string) (content []byte, sha string, err error) + PutContent(repo, path, ref, sha, message string, content []byte) error +} + +// ConflictError reports an optimistic-lock (HTTP 409) failure from the Contents +// API, the signal CommitWithRetry retries on. Clients wrap the API's 409 in this +// type so the retry loop does not have to string-match raw gh output. +type ConflictError struct { + // Err is the underlying transport or CLI error, surfaced when the retry + // bound is exhausted. + Err error +} + +func (e *ConflictError) Error() string { + return fmt.Sprintf("contents API conflict (409): %v", e.Err) +} + +// Unwrap exposes the wrapped error to errors.Is/As. +func (e *ConflictError) Unwrap() error { return e.Err } + +// IsConflict reports whether err is (or wraps) a Contents API 409 conflict. It +// recognizes both the typed ConflictError and the raw gh-CLI 409 body, which +// carries "does not match" and a "409" status, so a client that forwards the +// gh error verbatim still triggers a retry. +func IsConflict(err error) bool { + if err == nil { + return false + } + var ce *ConflictError + if asConflict(err, &ce) { + return true + } + msg := err.Error() + return strings.Contains(msg, "does not match") && strings.Contains(msg, "409") +} + +// asConflict is a tiny errors.As wrapper kept local so the package has no hard +// dependency surface beyond the standard library at its call sites. +func asConflict(err error, target **ConflictError) bool { + for err != nil { + if ce, ok := err.(*ConflictError); ok { //nolint:errorlint // walked manually below + *target = ce + return true + } + u, ok := err.(interface{ Unwrap() error }) + if !ok { + return false + } + err = u.Unwrap() + } + return false +} + +// Mutate applies a caller's state change to the current manifest bytes and +// returns the new bytes to write. It MUST be re-appliable: CommitWithRetry calls +// it again after re-fetching the manifest on a 409, so it must derive the new +// bytes purely from the current bytes it is handed (for example, parse them, +// set only this env's ci.state. keys, and re-marshal) rather than from a +// stale in-memory snapshot. Re-fetching picks up the other writer's committed +// keys, and re-applying preserves both. +type Mutate func(current []byte) ([]byte, error) + +// Options carries the inputs CommitWithRetry needs. Required identity fields are +// explicit; Sleep is optional and defaults to time.Sleep. +type Options struct { + // Client performs the Contents API get/put. Required. + Client ContentsClient + // Repo is the "owner/name" repository slug. Required. + Repo string + // Path is the repo-relative manifest path. Required. + Path string + // Ref is the branch the write targets (the trunk branch). Required. + Ref string + // Message is the commit message for the write. Required. + Message string + // Mutate derives the bytes to write from the current manifest bytes. It is + // re-applied on every retry. Required. + Mutate Mutate + // Sleep is called between retries. Defaults to time.Sleep; tests inject a + // no-op so no real time passes. + Sleep func(time.Duration) +} + +// CommitWithRetry performs an optimistic-locked read-modify-write of the +// manifest at opts.Ref. It fetches the current manifest and blob SHA, applies +// opts.Mutate to the current bytes, and PUTs with that SHA. On a 409 conflict it +// re-fetches, re-applies the mutation on top of the now-current manifest, and +// re-PUTs, up to maxAttempts with a staggered backoff. It returns the last +// conflict (or any non-409 error) when the bound is exhausted, so a genuinely +// stuck write still surfaces. +// +// Because Mutate is re-applied against the freshly fetched manifest, two +// finalize jobs that each set only their own env's ci.state. keys merge: +// the loser re-reads the winner's committed keys and re-applies its own on top, +// so the final manifest carries both. +func CommitWithRetry(opts Options) error { + sleep := opts.Sleep + if sleep == nil { + sleep = time.Sleep + } + + var lastErr error + for attempt := 1; attempt <= maxAttempts; attempt++ { + current, sha, err := opts.Client.GetContent(opts.Repo, opts.Path, opts.Ref) + if err != nil { + return fmt.Errorf("reading current manifest for state write: %w", err) + } + + next, err := opts.Mutate(current) + if err != nil { + return fmt.Errorf("applying state mutation: %w", err) + } + + err = opts.Client.PutContent(opts.Repo, opts.Path, opts.Ref, sha, opts.Message, next) + if err == nil { + return nil + } + if !IsConflict(err) { + return fmt.Errorf("state write via API failed: %w", err) + } + + // Optimistic-lock conflict: another writer committed between our read + // and our PUT. Re-fetch, re-apply, and retry so both writers' state + // merges rather than one being dropped. + lastErr = err + if attempt < maxAttempts { + sleep(time.Duration(attempt) * retryBackoff) + } + } + + return fmt.Errorf("state write via API still conflicting after %d attempts: %w", maxAttempts, lastErr) +} diff --git a/internal/statewrite/apiwrite_test.go b/internal/statewrite/apiwrite_test.go new file mode 100644 index 00000000..4df5d0ca --- /dev/null +++ b/internal/statewrite/apiwrite_test.go @@ -0,0 +1,188 @@ +package statewrite + +import ( + "errors" + "fmt" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// fakeContents is a stub ContentsClient that models the manifest as a single +// string and lets a test script which PUTs return a 409. It records each PUT so +// tests can assert the writer re-fetched and re-applied. +type fakeContents struct { + content string // current manifest on the "branch" + sha string // current blob SHA + + // putErrs is consumed one entry per PutContent call: a non-nil entry forces + // that PUT to fail without mutating state; nil applies the write. Extra PUTs + // past the slice default to applying successfully. + putErrs []error + + puts int // number of PutContent calls + gets int // number of GetContent calls + putSeen []string // content bytes presented to each PutContent + shaSeen []string // sha presented to each PutContent +} + +func (f *fakeContents) GetContent(_, _, _ string) ([]byte, string, error) { + f.gets++ + return []byte(f.content), f.sha, nil +} + +func (f *fakeContents) PutContent(_, _, _, sha, _ string, content []byte) error { + f.puts++ + f.putSeen = append(f.putSeen, string(content)) + f.shaSeen = append(f.shaSeen, sha) + if f.puts-1 < len(f.putErrs) { + if err := f.putErrs[f.puts-1]; err != nil { + return err + } + } + // Apply the write: advance the stored content and bump the blob SHA so a + // re-fetch sees the new state under a new optimistic-lock token. + f.content = string(content) + f.sha = fmt.Sprintf("%s-next", sha) + return nil +} + +// appendLine is a re-appliable mutation: it adds a line for one env, idempotently +// (it never duplicates a line it already added), so re-applying on top of another +// writer's committed line preserves both. This mirrors a finalize that sets only +// its own ci.state. keys. +func appendLine(line string) Mutate { + return func(current []byte) ([]byte, error) { + body := string(current) + if strings.Contains(body, line) { + return current, nil + } + if body != "" && !strings.HasSuffix(body, "\n") { + body += "\n" + } + return []byte(body + line + "\n"), nil + } +} + +// rawConflict mirrors the raw gh-CLI 409 body so IsConflict's string path is +// exercised, wrapped in the typed ConflictError clients are expected to return. +func rawConflict() error { + return &ConflictError{Err: errors.New(`{"message":".github/manifest.yaml does not match abc123","status":"409"}`)} +} + +// noSleep is an injected sleep that records that it was called but lets no real +// time pass, keeping the test fast and deterministic. +func noSleep(calls *int) func(time.Duration) { + return func(time.Duration) { *calls++ } +} + +func TestCommitWithRetry_RetriesOn409AndMergesConcurrentWrite(t *testing.T) { + // Env A has already committed its state line; the branch carries it. Our + // caller (env B) sets its own line. The first PUT loses the optimistic-lock + // race (409); the re-fetch must pick up env A's line and the re-apply must + // preserve it while adding env B's, so the final manifest carries BOTH. + fake := &fakeContents{ + content: "ci.state.test: A\n", + sha: "sha-0", + putErrs: []error{rawConflict()}, // first PUT 409s, second succeeds + } + + var slept int + start := time.Now() + err := CommitWithRetry(Options{ + Client: fake, + Repo: "owner/name", + Path: ".github/manifest.yaml", + Ref: "main", + Message: "chore: record state on staging", + Mutate: appendLine("ci.state.staging: B"), + Sleep: noSleep(&slept), + }) + + require.NoError(t, err) + // Re-fetched after the 409: two GETs, two PUTs. + assert.Equal(t, 2, fake.gets, "writer must re-fetch the manifest after a 409") + assert.Equal(t, 2, fake.puts, "writer must re-PUT after a 409") + assert.Equal(t, 1, slept, "writer must back off once between the two attempts") + // Merge semantics: the final committed manifest carries BOTH env A's and + // env B's state, proving the mutation was re-applied on top of the winner. + assert.Contains(t, fake.content, "ci.state.test: A", "the other writer's state must survive") + assert.Contains(t, fake.content, "ci.state.staging: B", "this caller's state must be written") + // No real time elapsed: the injected sleep is a no-op. + assert.Less(t, time.Since(start), time.Second, "injected sleep must let no real time pass") +} + +func TestCommitWithRetry_ErrorsAfterBoundedAttempts(t *testing.T) { + // Every PUT 409s: the writer must exhaust its bound and surface the conflict + // rather than spinning forever or swallowing the error. + errs := make([]error, maxAttempts) + for i := range errs { + errs[i] = rawConflict() + } + fake := &fakeContents{content: "ci.state.test: A\n", sha: "sha-0", putErrs: errs} + + var slept int + err := CommitWithRetry(Options{ + Client: fake, + Repo: "owner/name", + Path: ".github/manifest.yaml", + Ref: "main", + Message: "chore: record state", + Mutate: appendLine("ci.state.staging: B"), + Sleep: noSleep(&slept), + }) + + require.Error(t, err) + assert.Contains(t, err.Error(), "still conflicting") + assert.True(t, IsConflict(err), "the surfaced error must remain recognizable as a 409") + assert.Equal(t, maxAttempts, fake.puts, "writer must try exactly the bounded number of times") + assert.Equal(t, maxAttempts-1, slept, "writer must back off between attempts but not after the last") +} + +func TestCommitWithRetry_NonConflictErrorIsNotRetried(t *testing.T) { + // A non-409 error (e.g. auth) must surface immediately without retrying. + fake := &fakeContents{ + content: "ci.state.test: A\n", + sha: "sha-0", + putErrs: []error{errors.New("HTTP 401: bad credentials")}, + } + + var slept int + err := CommitWithRetry(Options{ + Client: fake, + Repo: "owner/name", + Path: ".github/manifest.yaml", + Ref: "main", + Message: "chore: record state", + Mutate: appendLine("ci.state.staging: B"), + Sleep: noSleep(&slept), + }) + + require.Error(t, err) + assert.False(t, IsConflict(err)) + assert.Equal(t, 1, fake.puts, "a non-409 error must not be retried") + assert.Equal(t, 0, slept, "a non-409 error must not back off") +} + +func TestIsConflict(t *testing.T) { + tests := []struct { + name string + err error + want bool + }{ + {"nil", nil, false}, + {"typed conflict", &ConflictError{Err: errors.New("boom")}, true}, + {"raw gh 409 body", errors.New(`{"message":"manifest.yaml does not match abc","status":"409"}`), true}, + {"wrapped typed conflict", fmt.Errorf("committing state: %w", &ConflictError{Err: errors.New("boom")}), true}, + {"unrelated error", errors.New("HTTP 500"), false}, + {"does-not-match without 409", errors.New("ref does not match"), false}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, tc.want, IsConflict(tc.err)) + }) + } +} diff --git a/internal/statewrite/ghclient.go b/internal/statewrite/ghclient.go new file mode 100644 index 00000000..24700332 --- /dev/null +++ b/internal/statewrite/ghclient.go @@ -0,0 +1,76 @@ +package statewrite + +import ( + "encoding/base64" + "fmt" + "os/exec" + "strings" +) + +// ghContents is the production ContentsClient. It shells out to the gh CLI to +// read and write manifest blobs through the GitHub Contents REST API, producing +// signed (Verified) commits that, with a bypass-capable token, can update a +// protected trunk branch. +type ghContents struct{} + +// NewGHClient returns the production ContentsClient backed by the gh CLI. +func NewGHClient() ContentsClient { + return ghContents{} +} + +// GetContent returns the current manifest bytes and blob SHA at ref. It fetches +// the raw file and the blob SHA in two gh calls. A file that does not yet exist +// (either gh call errors, or the SHA is empty) returns nil bytes, an empty SHA, +// and a nil error so the first write creates it rather than failing. +func (ghContents) GetContent(repo, path, ref string) ([]byte, string, error) { + apiPath := fmt.Sprintf("repos/%s/contents/%s?ref=%s", repo, path, ref) + + raw, err := exec.Command("gh", "api", apiPath, "-H", "Accept: application/vnd.github.raw").Output() + if err != nil { + return nil, "", nil + } + + shaOut, err := exec.Command("gh", "api", apiPath, "--jq", ".sha").Output() + if err != nil { + return nil, "", nil + } + sha := strings.TrimSpace(string(shaOut)) + if sha == "" { + return nil, "", nil + } + return raw, sha, nil +} + +// PutContent writes content at ref through the Contents API. When sha is +// non-empty the write is an update guarded by that optimistic-lock token; an +// empty sha creates the file. It classifies a 409 optimistic-lock failure as a +// ConflictError so the retry loop recognizes it. +func (ghContents) PutContent(repo, path, ref, sha, message string, content []byte) error { + b64 := base64.StdEncoding.EncodeToString(content) + args := []string{ + "api", fmt.Sprintf("repos/%s/contents/%s", repo, path), "-X", "PUT", + "-f", "message=" + message, + "-f", "content=" + b64, + "-f", "branch=" + ref, + } + if sha != "" { + args = append(args, "-f", "sha="+sha) + } + out, err := exec.Command("gh", args...).CombinedOutput() + return classifyPutError(string(out), err) +} + +// classifyPutError maps a gh PUT result to a typed error. A nil err is success. +// A 409 optimistic-lock failure (the body carries "does not match" alongside a +// 409, "Conflict", or "is at" marker) becomes a ConflictError so the retry loop +// re-fetches and re-applies; any other failure is wrapped verbatim. +func classifyPutError(out string, err error) error { + if err == nil { + return nil + } + if strings.Contains(out, "does not match") && + (strings.Contains(out, "409") || strings.Contains(out, "Conflict") || strings.Contains(out, "is at")) { + return &ConflictError{Err: fmt.Errorf("%s: %w", strings.TrimSpace(out), err)} + } + return fmt.Errorf("%s: %w", strings.TrimSpace(out), err) +} diff --git a/internal/statewrite/ghclient_test.go b/internal/statewrite/ghclient_test.go new file mode 100644 index 00000000..5ca5e4d3 --- /dev/null +++ b/internal/statewrite/ghclient_test.go @@ -0,0 +1,58 @@ +package statewrite + +import ( + "errors" + "testing" +) + +func TestClassifyPutError(t *testing.T) { + tests := []struct { + name string + out string + err error + wantConflict bool + wantNil bool + }{ + { + name: "409 conflict", + out: `{"message":"manifest.yaml does not match abc","status":"409"}`, + err: errors.New("exit 1"), + wantConflict: true, + }, + { + name: "conflict word", + out: `manifest.yaml does not match abc: Conflict`, + err: errors.New("exit 1"), + wantConflict: true, + }, + { + name: "401 not conflict", + out: "401 Unauthorized", + err: errors.New("exit 1"), + wantConflict: false, + }, + { + name: "nil err", + out: "", + err: nil, + wantNil: true, + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + got := classifyPutError(tc.out, tc.err) + if tc.wantNil { + if got != nil { + t.Fatalf("classifyPutError() = %v, want nil", got) + } + return + } + if got == nil { + t.Fatal("classifyPutError() = nil, want non-nil error") + } + if IsConflict(got) != tc.wantConflict { + t.Errorf("IsConflict() = %v, want %v", IsConflict(got), tc.wantConflict) + } + }) + } +}