diff --git a/pkg/schema/mysql/apply_comments.sql b/pkg/schema/mysql/apply_comments.sql index 7e771a780..81c5b1129 100644 --- a/pkg/schema/mysql/apply_comments.sql +++ b/pkg/schema/mysql/apply_comments.sql @@ -6,6 +6,8 @@ CREATE TABLE `apply_comments` ( `posted_volume` int DEFAULT NULL, `posted_phase` varchar(32) DEFAULT NULL, `pending_freeze_github_comment_id` bigint DEFAULT NULL, + `observer_owner` varchar(255) DEFAULT NULL, + `observer_heartbeat_at` datetime DEFAULT NULL, `edit_count` int NOT NULL DEFAULT '0', `last_edited_at` datetime DEFAULT NULL, `superseded_at` datetime DEFAULT NULL, diff --git a/pkg/schema/postgres/apply_comments.sql b/pkg/schema/postgres/apply_comments.sql index e0c863cce..a7ba28f76 100644 --- a/pkg/schema/postgres/apply_comments.sql +++ b/pkg/schema/postgres/apply_comments.sql @@ -6,6 +6,8 @@ CREATE TABLE apply_comments ( posted_volume integer DEFAULT NULL, posted_phase varchar(32) DEFAULT NULL, pending_freeze_github_comment_id bigint DEFAULT NULL, + observer_owner varchar(255) DEFAULT NULL, + observer_heartbeat_at timestamp DEFAULT NULL, edit_count integer NOT NULL DEFAULT 0, last_edited_at timestamp DEFAULT NULL, superseded_at timestamp DEFAULT NULL, diff --git a/pkg/storage/internal/sqlstore/apply_comments.go b/pkg/storage/internal/sqlstore/apply_comments.go index 0184db191..36cc94b7b 100644 --- a/pkg/storage/internal/sqlstore/apply_comments.go +++ b/pkg/storage/internal/sqlstore/apply_comments.go @@ -322,6 +322,54 @@ func (s *applyCommentStore) ReclaimStaleSummaryClaim(ctx context.Context, applyI return rows == 1, nil } +// ClaimProgressCommentAuthority claims — or renews for its current holder — +// the durable authority to edit an apply's tracked progress comment while the +// parent apply lease is legitimately unheld. It is a single conditional update +// on the tracked progress comment row, so concurrent claimants race on the row +// lock and exactly one sees the affected row: the update lands when the row +// records no observer, when the caller already holds the authority (a +// renewal, which also refreshes the heartbeat), or when the recorded holder's +// heartbeat is older than storage.ProgressCommentAuthorityStaleAfter (a +// crashed holder being taken over). Zero affected rows means another observer +// holds a fresh authority or no tracked progress comment row exists — either +// way the caller must skip its GitHub side effect. Deliberately +// lease-agnostic, mirroring ClaimSummaryComment: the claim is the authority. +func (s *applyCommentStore) ClaimProgressCommentAuthority(ctx context.Context, applyID int64, owner string) (bool, error) { + result, err := s.db.ExecContext(ctx, ` + UPDATE apply_comments + SET observer_owner = ?, observer_heartbeat_at = NOW(), updated_at = NOW() + WHERE apply_id = ? AND comment_state = ? + AND (observer_owner IS NULL + OR observer_owner = ? + OR observer_heartbeat_at < `+s.dialect.RelativeTime(TimestampPrecisionDefault, BeforeCurrentTime, ParameterIntervalAmount(), IntervalSecond)+`) + `, owner, applyID, state.Comment.Progress, owner, int64(storage.ProgressCommentAuthorityStaleAfter.Seconds())) + if err != nil { + return false, fmt.Errorf("claim progress-comment authority for apply %d owner %s: %w", applyID, owner, err) + } + rows, err := result.RowsAffected() + if err != nil { + return false, fmt.Errorf("read progress-comment authority claim rows affected for apply %d owner %s: %w", applyID, owner, err) + } + if rows == 1 { + return true, nil + } + // MySQL reports rows changed, not rows matched: a holder renewing inside + // the heartbeat column's timestamp granularity writes identical values and + // reports zero rows. Distinguish that held-by-the-caller no-op from a lost + // or unclaimable authority by re-reading the recorded owner. + var current sql.NullString + err = s.db.QueryRowContext(ctx, ` + SELECT observer_owner FROM apply_comments WHERE apply_id = ? AND comment_state = ? + `, applyID, state.Comment.Progress).Scan(¤t) + if errors.Is(err, sql.ErrNoRows) { + return false, nil + } + if err != nil { + return false, fmt.Errorf("verify progress-comment authority for apply %d owner %s: %w", applyID, owner, err) + } + return current.Valid && current.String == owner, nil +} + // ReleaseSummaryClaim deletes the summary claim sentinel for an apply so a // later publisher can retry without waiting out the stale-claim window. Only // the sentinel form (github_comment_id = 0) is deleted — a marker that already diff --git a/pkg/storage/internal/sqlstore/apply_comments_test.go b/pkg/storage/internal/sqlstore/apply_comments_test.go index cf1e375ef..55308138e 100644 --- a/pkg/storage/internal/sqlstore/apply_comments_test.go +++ b/pkg/storage/internal/sqlstore/apply_comments_test.go @@ -66,6 +66,92 @@ func TestApplyCommentStore_ReclaimStaleSummaryClaim(t *testing.T) { assert.False(t, reclaimed, "a posted summary must never be reclaimed") } +// TestApplyCommentStore_ReleaseSummaryClaim verifies release deletes only the +// sentinel form of the summary marker: a released claim can be re-won, a +// missing marker releases without error, and a marker recording a real posted +// comment survives release untouched. +func TestApplyCommentStore_ReleaseSummaryClaim(t *testing.T) { + clearTables(t) + ctx := t.Context() + store := NewMySQL(testDB) + + lock := createTestLock(t, store, "testdb", "mysql") + apply := createTestApply(t, store, lock, "apply_comment_release", 1) + + require.NoError(t, store.ApplyComments().ReleaseSummaryClaim(ctx, apply.ID), "releasing a missing claim is not an error") + + won, err := store.ApplyComments().ClaimSummaryComment(ctx, apply.ID) + require.NoError(t, err) + require.True(t, won) + require.NoError(t, store.ApplyComments().ReleaseSummaryClaim(ctx, apply.ID)) + + won, err = store.ApplyComments().ClaimSummaryComment(ctx, apply.ID) + require.NoError(t, err) + assert.True(t, won, "a released claim must be re-winnable") + + // Convert the claim to a posted summary; release must not delete it. + require.NoError(t, store.ApplyComments().Upsert(ctx, &storage.ApplyComment{ + ApplyID: apply.ID, CommentState: state.Comment.Summary, GitHubCommentID: 9001, + })) + require.NoError(t, store.ApplyComments().ReleaseSummaryClaim(ctx, apply.ID)) + posted, err := store.ApplyComments().Get(ctx, apply.ID, state.Comment.Summary) + require.NoError(t, err) + require.NotNil(t, posted, "a recorded posted summary must survive release") + assert.Equal(t, int64(9001), posted.GitHubCommentID) +} + +// TestApplyCommentStore_ClaimProgressCommentAuthority verifies the +// crashed-holder handover of the progress-comment authority: a recorded owner +// whose heartbeat is older than the stale window transfers to the next +// claimant, exactly once — the takeover stamps a fresh heartbeat, so a third +// claimant loses again. Aging the heartbeat requires backdating +// observer_heartbeat_at with raw SQL, which the storage interface cannot +// express, so the scenario lives in each dialect suite; the fresh-row claim +// decisions run on both dialects through the parity suite. +func TestApplyCommentStore_ClaimProgressCommentAuthority(t *testing.T) { + clearTables(t) + ctx := t.Context() + store := NewMySQL(testDB) + + lock := createTestLock(t, store, "testdb", "mysql") + apply := createTestApply(t, store, lock, "apply_comment_authority", 1) + + require.NoError(t, store.ApplyComments().Upsert(ctx, &storage.ApplyComment{ + ApplyID: apply.ID, CommentState: state.Comment.Progress, GitHubCommentID: 555, + })) + + held, err := store.ApplyComments().ClaimProgressCommentAuthority(ctx, apply.ID, "pod-a/1/comment-observer") + require.NoError(t, err) + require.True(t, held, "first claim on an unowned progress comment must win") + + held, err = store.ApplyComments().ClaimProgressCommentAuthority(ctx, apply.ID, "pod-b/2/comment-observer") + require.NoError(t, err) + require.False(t, held, "a second owner must lose while the holder's heartbeat is fresh") + + // A crashed holder hands over only after its heartbeat goes stale, and + // exactly one successor wins the handover. + backdateProgressObserverHeartbeat(t, apply.ID) + held, err = store.ApplyComments().ClaimProgressCommentAuthority(ctx, apply.ID, "pod-b/2/comment-observer") + require.NoError(t, err) + assert.True(t, held, "a stale authority transfers to the next claimant") + + held, err = store.ApplyComments().ClaimProgressCommentAuthority(ctx, apply.ID, "pod-c/3/comment-observer") + require.NoError(t, err) + assert.False(t, held, "a just-transferred authority is fresh again; a third owner loses") +} + +// backdateProgressObserverHeartbeat pushes an apply's progress-comment +// authority heartbeat past the stale window, simulating an observer that +// stopped renewing (crashed pod or cleared observer). +func backdateProgressObserverHeartbeat(t *testing.T, applyID int64) { + t.Helper() + _, err := testDB.ExecContext(t.Context(), ` + UPDATE apply_comments SET observer_heartbeat_at = NOW() - INTERVAL ? SECOND + WHERE apply_id = ? AND comment_state = ? + `, int64(storage.ProgressCommentAuthorityStaleAfter.Seconds())+1, applyID, state.Comment.Progress) + require.NoError(t, err) +} + // backdateSummaryClaim pushes an apply's summary marker updated_at past the // stale-claim window, simulating a publisher that crashed after claiming. func backdateSummaryClaim(t *testing.T, applyID int64) { diff --git a/pkg/storage/internal/sqlstore/postgres_integration_test.go b/pkg/storage/internal/sqlstore/postgres_integration_test.go index 6da47775d..2c6839d89 100644 --- a/pkg/storage/internal/sqlstore/postgres_integration_test.go +++ b/pkg/storage/internal/sqlstore/postgres_integration_test.go @@ -58,6 +58,57 @@ func TestPostgresStorageParity(t *testing.T) { t.Run("ApplyCommentReclaimStaleSummaryClaim", func(t *testing.T) { testPostgresApplyCommentReclaimStaleSummaryClaim(t, h) }) t.Run("ApplyCommentMutationsStampUpdatedAt", func(t *testing.T) { testPostgresApplyCommentMutationsStampUpdatedAt(t, h) }) t.Run("ApplyCommentClaimConversionRestartsStaleWindow", func(t *testing.T) { testPostgresApplyCommentClaimConversionRestartsStaleWindow(t, h) }) + t.Run("ApplyCommentProgressAuthorityStaleTakeover", func(t *testing.T) { testPostgresApplyCommentProgressAuthorityStaleTakeover(t, h) }) +} + +// backdatePostgresProgressObserverHeartbeat pushes an apply's progress-comment +// authority heartbeat past the stale window, simulating an observer that +// stopped renewing (crashed pod or cleared observer). +func backdatePostgresProgressObserverHeartbeat(t *testing.T, db *sql.DB, applyID int64) { + t.Helper() + _, err := db.ExecContext(t.Context(), ` + UPDATE apply_comments SET observer_heartbeat_at = now() - make_interval(secs => $1) + WHERE apply_id = $2 AND comment_state = $3 + `, int64(storage.ProgressCommentAuthorityStaleAfter.Seconds())+1, applyID, state.Comment.Progress) + require.NoError(t, err) +} + +// testPostgresApplyCommentProgressAuthorityStaleTakeover verifies the +// crashed-holder handover of the progress-comment authority on PostgreSQL: a +// recorded owner whose heartbeat is older than the stale window transfers to +// the next claimant, exactly once — the takeover stamps a fresh heartbeat, so +// a third claimant loses again. Aging the heartbeat requires backdating +// observer_heartbeat_at with raw SQL, which the storage interface cannot +// express, so the scenario lives in each dialect suite; the fresh-row claim +// decisions run on both dialects through the parity suite. +func testPostgresApplyCommentProgressAuthorityStaleTakeover(t *testing.T, h postgresHarness) { + store := h.NewStorage(t) + ctx := t.Context() + + lock := storagetest.CreateLock(t, store, "comment_authority_stale_db", storage.DatabaseTypeMySQL) + apply := storagetest.CreateApply(t, store, lock, "apply_comment_authority_stale", 723) + + require.NoError(t, store.ApplyComments().Upsert(ctx, &storage.ApplyComment{ + ApplyID: apply.ID, CommentState: state.Comment.Progress, GitHubCommentID: 555, + })) + + held, err := store.ApplyComments().ClaimProgressCommentAuthority(ctx, apply.ID, "pod-a/1/comment-observer") + require.NoError(t, err) + require.True(t, held, "first claim on an unowned progress comment must win") + + held, err = store.ApplyComments().ClaimProgressCommentAuthority(ctx, apply.ID, "pod-b/2/comment-observer") + require.NoError(t, err) + require.False(t, held, "a second owner must lose while the holder's heartbeat is fresh") + + backdatePostgresProgressObserverHeartbeat(t, h.db, apply.ID) + + held, err = store.ApplyComments().ClaimProgressCommentAuthority(ctx, apply.ID, "pod-b/2/comment-observer") + require.NoError(t, err) + assert.True(t, held, "a stale authority transfers to the next claimant") + + held, err = store.ApplyComments().ClaimProgressCommentAuthority(ctx, apply.ID, "pod-c/3/comment-observer") + require.NoError(t, err) + assert.False(t, held, "a just-transferred authority is fresh again; a third owner loses") } // backdatePostgresSummaryClaim pushes an apply's summary marker updated_at diff --git a/pkg/storage/storage.go b/pkg/storage/storage.go index 26d8c247c..4e1881b23 100644 --- a/pkg/storage/storage.go +++ b/pkg/storage/storage.go @@ -986,8 +986,32 @@ type ApplyCommentStore interface { // claim window. Deletes only the sentinel form of the marker — a recorded // real comment is never released. ReleaseSummaryClaim(ctx context.Context, applyID int64) error + + // ClaimProgressCommentAuthority atomically claims — or renews for its + // current holder — the durable authority to edit the tracked progress + // comment of an apply whose parent apply lease is legitimately unheld + // because its work runs under operation leases. The claim is a conditional + // update on the tracked progress comment row: it succeeds when the row has + // no recorded observer, when the caller already holds it, or when the + // recorded observer's heartbeat is older than + // ProgressCommentAuthorityStaleAfter (a crashed holder). Exactly one of any + // set of concurrent claimants wins the same handover, so two observers can + // never both believe they own the comment. Returns true when the caller now + // holds the authority; false when another observer holds it or no tracked + // progress comment row exists to claim. Deliberately lease-agnostic — the + // claim itself is the authority, mirroring the terminal summary claim. + ClaimProgressCommentAuthority(ctx context.Context, applyID int64, owner string) (bool, error) } +// ProgressCommentAuthorityStaleAfter is how long the progress-comment +// authority may go without a renewal before its holder is considered gone and +// another observer may take the authority over. Holders renew on every +// admitted GitHub side effect (at least once per progress poll tick), so +// anything older than this window is a stopped observer, not a slow one. It +// matches the apply lease staleness bound so comment ownership hands over on +// the same clock as drive ownership. +const ProgressCommentAuthorityStaleAfter = ApplyLeaseStaleAfter + // SummaryClaimStaleAfter is how long a summary claim sentinel // (apply_comments row with github_comment_id = 0) may go without an update // before it is considered abandoned by a crashed publisher and becomes diff --git a/pkg/storage/storagetest/apply_comments.go b/pkg/storage/storagetest/apply_comments.go index de50fffcd..e9f5a5fbb 100644 --- a/pkg/storage/storagetest/apply_comments.go +++ b/pkg/storage/storagetest/apply_comments.go @@ -16,12 +16,13 @@ import ( // pending-freeze marker, the exactly-once summary-claim machinery, and the // apply-lease guard on every lease-guarded mutation. // -// Three claim behaviors can only be proven against an aged row and stay in -// the dialect suites, which can backdate updated_at directly: the -// stale-window takeover in ReclaimStaleSummaryClaim, its refusal to reclaim -// a marker recording a posted comment, and the assertion that every mutation -// renews updated_at (the claim machinery's freshness signal). The parity -// suite covers the reclaim decisions that do not require aging a row. +// Claim behaviors that can only be proven against an aged row stay in the +// dialect suites, which can backdate timestamps directly: the stale-window +// takeover in ReclaimStaleSummaryClaim, its refusal to reclaim a marker +// recording a posted comment, the assertion that every mutation renews +// updated_at (the claim machinery's freshness signal), and the stale-heartbeat +// takeover in ClaimProgressCommentAuthority. The parity suite covers the +// claim decisions that do not require aging a row. func TestApplyComments(t *testing.T, h Harness) { // Upsert_And_Get verifies the tracked-comment round trip: an insert // stores the posted level and control phase, a conflicting upsert for the @@ -377,6 +378,52 @@ func TestApplyComments(t *testing.T, h Harness) { assert.False(t, won, "a superseded claim sentinel is not reclaimable") }) + // ClaimProgressCommentAuthority verifies the durable progress-comment edit + // authority's fresh-row decisions on every dialect: an apply with no + // tracked progress comment row has nothing to claim, the first claim on an + // unowned row wins and records its owner, the holder renews its own + // authority (on MySQL an identical-value renewal reports zero changed rows + // and must still be recognized as held), a second owner loses while the + // holder's heartbeat is fresh, and claims for different applies are + // independent. The stale-heartbeat takeover needs an aged row and lives in + // the dialect suites. + t.Run("ClaimProgressCommentAuthority", func(t *testing.T) { + ctx := t.Context() + store := h.NewStorage(t) + + lock := CreateLock(t, store, "comment_authority_db", storage.DatabaseTypeMySQL) + apply := CreateApply(t, store, lock, "apply_comment_authority", 713) + otherLock := CreateLock(t, store, "comment_authority_other_db", storage.DatabaseTypeMySQL) + other := CreateApply(t, store, otherLock, "apply_comment_authority_other", 714) + + held, err := store.ApplyComments().ClaimProgressCommentAuthority(ctx, apply.ID, "pod-a/1/comment-observer") + require.NoError(t, err) + assert.False(t, held, "no tracked progress comment row means nothing to claim") + + require.NoError(t, store.ApplyComments().Upsert(ctx, &storage.ApplyComment{ + ApplyID: apply.ID, CommentState: state.Comment.Progress, GitHubCommentID: 555, + })) + require.NoError(t, store.ApplyComments().Upsert(ctx, &storage.ApplyComment{ + ApplyID: other.ID, CommentState: state.Comment.Progress, GitHubCommentID: 556, + })) + + held, err = store.ApplyComments().ClaimProgressCommentAuthority(ctx, apply.ID, "pod-a/1/comment-observer") + require.NoError(t, err) + assert.True(t, held, "first claim on an unowned progress comment must win") + + held, err = store.ApplyComments().ClaimProgressCommentAuthority(ctx, apply.ID, "pod-a/1/comment-observer") + require.NoError(t, err) + assert.True(t, held, "the holder renews its own authority, including an identical-value renewal") + + held, err = store.ApplyComments().ClaimProgressCommentAuthority(ctx, apply.ID, "pod-b/2/comment-observer") + require.NoError(t, err) + assert.False(t, held, "a second owner must lose while the holder's heartbeat is fresh") + + held, err = store.ApplyComments().ClaimProgressCommentAuthority(ctx, other.ID, "pod-b/2/comment-observer") + require.NoError(t, err) + assert.True(t, held, "authorities for different applies are independent") + }) + // ReclaimStaleSummaryClaim_RequiresStaleSentinel verifies the reclaim // refusals that do not depend on aging a row: a missing marker is not // reclaimable, and a fresh sentinel is an in-flight publish and stays @@ -580,6 +627,12 @@ func TestApplyComments(t *testing.T, h Harness) { require.Error(t, err) }) + t.Run("ClaimProgressCommentAuthority_DBError", func(t *testing.T) { + store := h.NewUnreachableStorage(t) + _, err := store.ApplyComments().ClaimProgressCommentAuthority(t.Context(), 1, "pod-a/1/comment-observer") + require.Error(t, err) + }) + t.Run("ReleaseSummaryClaim_DBError", func(t *testing.T) { store := h.NewUnreachableStorage(t) require.Error(t, store.ApplyComments().ReleaseSummaryClaim(t.Context(), 1)) diff --git a/pkg/webhook/comment_authority_integration_test.go b/pkg/webhook/comment_authority_integration_test.go new file mode 100644 index 000000000..44649de0b --- /dev/null +++ b/pkg/webhook/comment_authority_integration_test.go @@ -0,0 +1,360 @@ +//go:build integration + +package webhook + +import ( + "database/sql" + "fmt" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/block/spirit/pkg/utils" + + "github.com/block/schemabot/pkg/state" + "github.com/block/schemabot/pkg/storage" + "github.com/block/schemabot/pkg/storage/mysqlstore" +) + +// operationScopedApplyFixture is one seeded apply whose work runs under +// operation leases: two keyed non-terminal operation rows, no lease on the +// parent applies row, and a tracked progress comment ready to edit. +type operationScopedApplyFixture struct { + st storage.Storage + db *sql.DB + apply *storage.Apply + tasks []*storage.Task + progressCommentID int64 +} + +// seedOperationScopedApply creates a running apply with keyed operation rows +// and a tracked progress comment, leaving the parent applies row's lease +// fields empty — the shape an operation-scoped rollout has between dispatch +// waves. repo scopes the rows so tests do not interfere with each other. +func seedOperationScopedApply(t *testing.T, repo, database string) *operationScopedApplyFixture { + t.Helper() + ctx := t.Context() + + schemabotDB, err := sql.Open("mysql", e2eSchemabotDSN) + require.NoError(t, err) + t.Cleanup(func() { utils.CloseAndLog(schemabotDB) }) + require.NoError(t, schemabotDB.PingContext(ctx)) + + st := mysqlstore.New(schemabotDB) + + lock := &storage.Lock{ + DatabaseName: database, + DatabaseType: "mysql", + Repository: repo, + PullRequest: 42, + Owner: repo + "#42", + } + require.NoError(t, st.Locks().Acquire(ctx, lock)) + lock, err = st.Locks().Get(ctx, database, "mysql") + require.NoError(t, err) + + apply := &storage.Apply{ + ApplyIdentifier: fmt.Sprintf("apply_authority_%s_%d", database, time.Now().UnixNano()), + LockID: lock.ID, + PlanID: 1, + Database: database, + DatabaseType: "mysql", + Repository: repo, + PullRequest: 42, + Environment: "staging", + InstallationID: 12345, + Engine: "spirit", + State: state.Apply.Running, + } + applyID, err := st.Applies().Create(ctx, apply) + require.NoError(t, err) + + // Two operation-keyed rows still copying under their own operation leases, + // while the parent applies row records no lease at all. + var firstOpID int64 + for i, key := range []string{"orders/-80", "orders/80-"} { + res, err := schemabotDB.ExecContext(ctx, ` + INSERT INTO apply_operations (apply_id, deployment, operation_key, state, lease_owner, lease_token) + VALUES (?, ?, ?, ?, ?, ?) + `, applyID, "primary", key, state.ApplyOperation.Running, + fmt.Sprintf("driver-host/%d/op", i), fmt.Sprintf("op-token-%d", i)) + require.NoError(t, err) + if i == 0 { + firstOpID, err = res.LastInsertId() + require.NoError(t, err) + } + } + + now := time.Now() + task := &storage.Task{ + TaskIdentifier: fmt.Sprintf("task_authority_%s_%d", database, now.UnixNano()), + ApplyID: applyID, + ApplyOperationID: &firstOpID, + PlanID: 1, + Database: database, + DatabaseType: "mysql", + Engine: "spirit", + Repository: repo, + PullRequest: 42, + Environment: "staging", + State: state.Task.Running, + TableName: "orders", + DDL: "ALTER TABLE orders ADD COLUMN region VARCHAR(32)", + DDLAction: "alter", + CreatedAt: now, + UpdatedAt: now, + } + _, err = st.Tasks().Create(ctx, task) + require.NoError(t, err) + + // The tracked progress comment the handler posted when the apply started. + progressCommentID := int64(700000 + time.Now().UnixNano()%100000) + require.NoError(t, st.ApplyComments().Upsert(ctx, &storage.ApplyComment{ + ApplyID: applyID, + CommentState: state.Comment.Progress, + GitHubCommentID: progressCommentID, + })) + + apply, err = st.Applies().Get(ctx, applyID) + require.NoError(t, err) + require.NotNil(t, apply) + require.Empty(t, apply.LeaseToken, "fixture apply must not hold a parent lease") + + tasks, err := st.Tasks().GetByApplyID(ctx, applyID) + require.NoError(t, err) + + return &operationScopedApplyFixture{ + st: st, + db: schemabotDB, + apply: apply, + tasks: tasks, + progressCommentID: progressCommentID, + } +} + +// progressObserverOwner reads the durable authority owner recorded on the +// apply's tracked progress comment row. +func progressObserverOwner(t *testing.T, db *sql.DB, applyID int64) sql.NullString { + t.Helper() + var owner sql.NullString + err := db.QueryRowContext(t.Context(), ` + SELECT observer_owner FROM apply_comments WHERE apply_id = ? AND comment_state = ? + `, applyID, state.Comment.Progress).Scan(&owner) + require.NoError(t, err) + return owner +} + +// requireNoGitHubCalls asserts a skipped observer produced no GitHub side +// effects. The observer callbacks are synchronous, so an empty channel after +// the callback returns means no call was made. +func requireNoGitHubCalls(t *testing.T, capture *commentCapture) { + t.Helper() + select { + case edit := <-capture.edits: + t.Fatalf("expected no GitHub comment edit, got edit of comment %d", edit.CommentID) + case created := <-capture.creates: + t.Fatalf("expected no GitHub comment create, got comment %d", created.ID) + default: + } +} + +// An apply whose operations run under operation leases holds the parent apply +// lease only transiently per dispatch wave. Between waves the observer must +// keep editing the PR progress comment — under the durable progress-comment +// authority rather than a lease — so operators watching the PR see a live +// rollout instead of a comment frozen mid-apply. +func TestObserverEditsProgressCommentForOperationScopedApply(t *testing.T) { + fx := seedOperationScopedApply(t, "org/authority-edit", "authority_edit_db") + + installClient, capture := setupFakeGitHubForComments(t) + capture.setBody(fx.progressCommentID, "seed progress body") + + obs := NewCommentObserver(CommentObserverConfig{ + GHClient: &fakeClientFactory{client: installClient}, + Storage: fx.st, + Repo: fx.apply.Repository, + PR: fx.apply.PullRequest, + InstallationID: fx.apply.InstallationID, + ApplyID: fx.apply.ID, + Logger: &capturingLogger{}, + }) + + obs.OnProgress(fx.apply, fx.tasks) + + select { + case edited := <-capture.edits: + assert.Equal(t, fx.progressCommentID, edited.CommentID) + assert.Contains(t, edited.Body, fx.apply.ApplyIdentifier, "progress edit must render the live apply") + assert.Contains(t, edited.Body, "running table copy", "progress edit must render the in-flight operations") + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for the progress comment edit") + } + + // The edit was performed under the durable authority recorded on the + // tracked comment row, owned by this process's observer identity. + owner := progressObserverOwner(t, fx.db, fx.apply.ID) + require.True(t, owner.Valid, "the observer must record its authority on the tracked comment row") + assert.Equal(t, storage.LeaseOwnerProcess()+"/comment-observer", owner.String) +} + +// Two observers polling the same operation-scoped apply from different +// processes must never both edit the progress comment: the first claims the +// durable authority, and the second loses the compare-and-swap, skips every +// GitHub side effect, and logs the skip with triage identifiers. +func TestConcurrentObserversShareOneProgressCommentAuthority(t *testing.T) { + fx := seedOperationScopedApply(t, "org/authority-race", "authority_race_db") + + installClient, capture := setupFakeGitHubForComments(t) + capture.setBody(fx.progressCommentID, "seed progress body") + + newObserver := func(owner string, logger *capturingLogger) *CommentObserver { + obs := NewCommentObserver(CommentObserverConfig{ + GHClient: &fakeClientFactory{client: installClient}, + Storage: fx.st, + Repo: fx.apply.Repository, + PR: fx.apply.PullRequest, + InstallationID: fx.apply.InstallationID, + ApplyID: fx.apply.ID, + Logger: logger, + }) + obs.authorityOwner = owner + return obs + } + + winner := newObserver("pod-a/1/comment-observer", &capturingLogger{}) + winner.OnProgress(fx.apply, fx.tasks) + + select { + case edited := <-capture.edits: + assert.Equal(t, fx.progressCommentID, edited.CommentID) + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for the winner's progress comment edit") + } + + loserLogger := &capturingLogger{} + loser := newObserver("pod-b/2/comment-observer", loserLogger) + loser.OnProgress(fx.apply, fx.tasks) + + requireNoGitHubCalls(t, capture) + + owner := progressObserverOwner(t, fx.db, fx.apply.ID) + require.True(t, owner.Valid) + assert.Equal(t, "pod-a/1/comment-observer", owner.String, "the loser must not take the authority over") + + var loggedSkip bool + for _, entry := range loserLogger.debugs { + if entry.msg != "observer: progress-comment authority not won (held by another observer, or no tracked comment row yet); skipping GitHub side effect" { + continue + } + loggedSkip = true + fields := fieldsOf(t, entry.args) + assert.Equal(t, fx.apply.ApplyIdentifier, fields["apply_id"]) + assert.Equal(t, fx.apply.Repository, fields["repo"]) + assert.Equal(t, fx.apply.PullRequest, fields["pr"]) + assert.Equal(t, "pod-b/2/comment-observer", fields["authority_owner"]) + } + assert.True(t, loggedSkip, "the losing observer must log its skipped edit with triage identifiers") +} + +// An apply whose driver holds the parent apply lease keeps the lease as the +// one authority for GitHub side effects: the lease holder edits without +// touching the durable comment authority, and an observer whose captured +// lease no longer matches the row skips even though operation-keyed work is +// in flight. +func TestLeaseHeldApplyKeepsLeaseAuthoritative(t *testing.T) { + fx := seedOperationScopedApply(t, "org/authority-lease", "authority_lease_db") + ctx := t.Context() + + // A driver claims the parent apply for a dispatch wave. + leaseAcquiredAt := time.Now() + _, err := fx.db.ExecContext(ctx, ` + UPDATE applies SET lease_owner = ?, lease_token = ?, lease_acquired_at = ? WHERE id = ? + `, "driver-host/9/dispatch", "wave-token", leaseAcquiredAt, fx.apply.ID) + require.NoError(t, err) + apply, err := fx.st.Applies().Get(ctx, fx.apply.ID) + require.NoError(t, err) + require.NotNil(t, apply) + + installClient, capture := setupFakeGitHubForComments(t) + capture.setBody(fx.progressCommentID, "seed progress body") + + newObserver := func(lease storage.ApplyLease, logger *capturingLogger) *CommentObserver { + return NewCommentObserver(CommentObserverConfig{ + GHClient: &fakeClientFactory{client: installClient}, + Storage: fx.st, + Repo: apply.Repository, + PR: apply.PullRequest, + InstallationID: apply.InstallationID, + ApplyID: apply.ID, + ApplyLease: lease, + Logger: logger, + }) + } + + holder := newObserver(apply.Lease(), &capturingLogger{}) + holder.OnProgress(apply, fx.tasks) + + select { + case edited := <-capture.edits: + assert.Equal(t, fx.progressCommentID, edited.CommentID) + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for the lease holder's progress comment edit") + } + + owner := progressObserverOwner(t, fx.db, fx.apply.ID) + assert.False(t, owner.Valid, "a lease-held apply must not record a progress-comment authority") + + // A stale observer whose captured lease was rotated away must skip while + // the row is held by another driver — the lease stays authoritative. + staleLogger := &capturingLogger{} + stale := newObserver(storage.ApplyLease{ApplyID: apply.ID, Owner: "old-driver", Token: "rotated-away"}, staleLogger) + stale.OnProgress(apply, fx.tasks) + + requireNoGitHubCalls(t, capture) + var loggedSkip bool + for _, entry := range staleLogger.errors { + if entry.msg == "observer: apply lease no longer owns apply; skipping GitHub side effect" { + loggedSkip = true + } + } + assert.True(t, loggedSkip, "the stale observer must log its skipped edit") +} + +// The observer's tick works from a poller snapshot that can predate a +// dispatch wave re-claiming the parent apply lease. The authority gate +// re-reads the apply row before claiming, so an observer holding a stale +// no-lease snapshot skips every GitHub side effect while the wave's lease +// holder owns the comment — and never records a durable authority alongside +// the live lease. +func TestParentLeaseReclaimAfterSnapshotDeniesAuthority(t *testing.T) { + fx := seedOperationScopedApply(t, "org/authority-reclaim", "authority_reclaim_db") + ctx := t.Context() + + installClient, capture := setupFakeGitHubForComments(t) + capture.setBody(fx.progressCommentID, "seed progress body") + + obs := NewCommentObserver(CommentObserverConfig{ + GHClient: &fakeClientFactory{client: installClient}, + Storage: fx.st, + Repo: fx.apply.Repository, + PR: fx.apply.PullRequest, + InstallationID: fx.apply.InstallationID, + ApplyID: fx.apply.ID, + Logger: &capturingLogger{}, + }) + + // A dispatch wave claims the parent apply after the observer's snapshot + // (fx.apply) was taken, so the snapshot still records no lease. + _, err := fx.db.ExecContext(ctx, ` + UPDATE applies SET lease_owner = ?, lease_token = ?, lease_acquired_at = ? WHERE id = ? + `, "driver-host/9/dispatch", "wave-token", time.Now(), fx.apply.ID) + require.NoError(t, err) + + obs.OnProgress(fx.apply, fx.tasks) + + requireNoGitHubCalls(t, capture) + owner := progressObserverOwner(t, fx.db, fx.apply.ID) + assert.False(t, owner.Valid, "no durable authority may be recorded while a driver holds the parent lease") +} diff --git a/pkg/webhook/comment_observer.go b/pkg/webhook/comment_observer.go index dd503c13b..862b24900 100644 --- a/pkg/webhook/comment_observer.go +++ b/pkg/webhook/comment_observer.go @@ -33,6 +33,7 @@ type CommentObserver struct { supportChannel api.SupportChannelConfig tenant string logger interface { + Debug(msg string, args ...any) Info(msg string, args ...any) Error(msg string, args ...any) } @@ -55,6 +56,15 @@ type CommentObserver struct { // apply-lease checks and lease-scoped storage writes accordingly. aggregateTerminalCASWinner bool + // authorityOwner identifies this process to the durable progress-comment + // authority claim (see ClaimProgressCommentAuthority). It is + // process-scoped, not observer-instance-scoped, so a replacement observer + // in the same process (e.g. after a re-registration) takes the authority + // over immediately instead of waiting out its predecessor's staleness + // window, while observers on other pods still hand over only through the + // claim. + authorityOwner string + mu sync.Mutex lastProgressPost time.Time lastState string @@ -98,6 +108,18 @@ type CommentObserver struct { // but whose tracking write failed, so later ticks adopt it (retry the write // with the known comment ID) instead of posting a duplicate. pendingRotation *pendingProgressRotation + + // authorityMu guards the per-callback memo of the durable progress-comment + // authority decision below. It is separate from mu because OnProgress holds + // mu for its entire tick while OnTerminal runs without it, and the gate is + // reached from inside both. + authorityMu sync.Mutex + // authorityDecided marks that the authority decision was already made + // during the current observer callback, so the gate's later invocations on + // the same callback reuse authorityHeld instead of re-reading storage and + // re-writing the claim row. Each callback starts with a fresh decision. + authorityDecided bool + authorityHeld bool } // pendingProgressRotation identifies a rotation progress comment that was @@ -146,6 +168,7 @@ type CommentObserverConfig struct { Tenant string Logger interface { + Debug(msg string, args ...any) Info(msg string, args ...any) Error(msg string, args ...any) } @@ -218,6 +241,7 @@ func NewCommentObserver(cfg CommentObserverConfig) *CommentObserver { logger: cfg.Logger, OnTerminalHook: cfg.OnTerminalHook, clock: clk, + authorityOwner: storage.LeaseOwnerProcess() + "/comment-observer", } } @@ -240,6 +264,7 @@ func NewAggregateTerminalCommentObserver(cfg CommentObserverConfig) *CommentObse func (o *CommentObserver) OnProgress(apply *storage.Apply, tasks []*storage.Task) { o.mu.Lock() defer o.mu.Unlock() + o.resetProgressCommentAuthorityDecision() if !o.leaseStillOwnsObserver(apply, "progress") { return } @@ -398,6 +423,7 @@ func (o *CommentObserver) OnProgress(apply *storage.Apply, tasks []*storage.Task // Edits the active comment to final state, posts summary comment, // and updates check runs. func (o *CommentObserver) OnTerminal(apply *storage.Apply, tasks []*storage.Task) { + o.resetProgressCommentAuthorityDecision() if !o.leaseStillOwnsObserver(apply, "terminal") { return } @@ -725,9 +751,11 @@ func (o *CommentObserver) leaseStillOwnsObserver(apply *storage.Apply, operation lease = apply.Lease() } if !lease.Valid() { - o.logError(apply, "observer: apply lease unavailable; skipping GitHub side effect", - "operation", operation) - return false + // No parent apply lease exists anywhere — not on this observer and not + // on the apply row. For an apply whose work runs under operation + // leases, that is the normal shape between dispatch waves, so the + // durable progress-comment authority decides instead of the lease. + return o.progressCommentAuthorityOwnsObserver(apply, operation) } // GitHub comments and check updates are side effects outside MySQL's @@ -737,6 +765,14 @@ func (o *CommentObserver) leaseStillOwnsObserver(apply *storage.Apply, operation ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) defer cancel() if err := o.stor.Applies().CheckLease(ctx, lease); err != nil { + if apply != nil && !apply.Lease().Valid() { + // This observer's construction-time lease no longer matches, and + // the apply row records no parent lease at all — the lease was + // released back (an operation-scoped drive holds the parent only + // transiently per dispatch wave), not claimed by a newer owner. + // The durable progress-comment authority decides instead. + return o.progressCommentAuthorityOwnsObserver(apply, operation) + } o.logError(apply, "observer: apply lease no longer owns apply; skipping GitHub side effect", "operation", operation, "lease_owner", lease.Owner, @@ -746,6 +782,163 @@ func (o *CommentObserver) leaseStillOwnsObserver(apply *storage.Apply, operation return true } +// progressCommentAuthorityOwnsObserver reports whether this observer may +// perform a GitHub side effect for an apply whose parent apply lease is +// legitimately unheld. An apply whose operations are dispatched under +// operation leases holds the parent lease only transiently per dispatch wave, +// so the progress comment would otherwise go silent for the whole rollout and +// operators would read a live apply as dead. The authority granted here is +// durable and cross-pod safe: a compare-and-swap ownership recorded on the +// tracked progress comment row (see ClaimProgressCommentAuthority), so among +// authority-path observers at most one at a time edits the comment and a +// crashed holder hands over only after its heartbeat goes stale. A +// lease-admitted observer is governed by the lease checks instead and never +// touches the recorded authority. It is granted only while operation-scoped +// work is in flight — an apply that holds (or should hold) a parent lease +// stays governed by the lease checks. +// +// The decision is made once per observer callback against freshly read +// storage rows — including a re-read of the parent lease columns, so a +// dispatch wave that re-claimed the parent since the poller's snapshot denies +// the authority — then reused by the callback's remaining side-effect checks. +// Claiming once per callback also renews the holder's heartbeat well inside +// its staleness window. +func (o *CommentObserver) progressCommentAuthorityOwnsObserver(apply *storage.Apply, operation string) bool { + if apply == nil { + o.logError(apply, "observer: apply lease unavailable and no apply loaded to resolve progress-comment authority; skipping GitHub side effect", + "operation", operation) + return false + } + o.authorityMu.Lock() + if o.authorityDecided { + held := o.authorityHeld + o.authorityMu.Unlock() + if !held { + o.logger.Debug("observer: progress-comment authority already denied this callback; skipping GitHub side effect", + append(apply.LogAttrs(), "operation", operation, "authority_owner", o.authorityOwner)...) + } + return held + } + o.authorityMu.Unlock() + + held := o.decideProgressCommentAuthority(apply, operation) + o.authorityMu.Lock() + o.authorityDecided, o.authorityHeld = true, held + o.authorityMu.Unlock() + return held +} + +// resetProgressCommentAuthorityDecision discards the previous callback's +// authority decision so the next gate invocation decides afresh. +func (o *CommentObserver) resetProgressCommentAuthorityDecision() { + o.authorityMu.Lock() + o.authorityDecided = false + o.authorityHeld = false + o.authorityMu.Unlock() +} + +// decideProgressCommentAuthority performs the storage reads and the claim +// behind progressCommentAuthorityOwnsObserver, in fail-closed order: no +// operation-scoped work in flight denies, a fresh parent-lease re-read +// showing a holder or a terminal apply denies, and only then is the durable +// claim attempted. +func (o *CommentObserver) decideProgressCommentAuthority(apply *storage.Apply, operation string) bool { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + inFlight, err := o.operationScopedWorkInFlight(ctx, apply) + if err != nil { + o.logger.Error("observer: failed to determine whether operation-scoped work is in flight; skipping GitHub side effect", + append(apply.LogAttrs(), "operation", operation, "error", err)...) + return false + } + if !inFlight { + o.logger.Error("observer: apply lease unavailable and no operation-scoped work in flight; skipping GitHub side effect", + append(apply.LogAttrs(), "operation", operation)...) + return false + } + + // The apply passed in is the poller's snapshot, up to a tick old. Re-read + // the row before claiming: a dispatch wave may have re-claimed the parent + // lease since the snapshot — its holder is governed by the lease checks + // and owns the comment while the lease lasts — or the projection may have + // settled the apply terminal, handing the comment to the terminal publish. + fresh, err := o.stor.Applies().Get(ctx, o.applyID) + if err != nil { + o.logger.Error("observer: failed to re-read the apply for the progress-comment authority; skipping GitHub side effect", + append(apply.LogAttrs(), "operation", operation, "error", err)...) + return false + } + if fresh == nil { + o.logger.Error("observer: apply row no longer exists for the progress-comment authority; skipping GitHub side effect", + append(apply.LogAttrs(), "operation", operation)...) + return false + } + if fresh.Lease().Valid() { + o.logger.Debug("observer: parent apply lease re-claimed since the poller snapshot; the lease holder owns the comment, skipping GitHub side effect", + append(apply.LogAttrs(), "operation", operation, "lease_owner", fresh.LeaseOwner)...) + return false + } + if state.IsTerminalApplyState(fresh.State) { + o.logger.Debug("observer: apply settled terminal since the poller snapshot; the terminal publish owns the comment, skipping GitHub side effect", + append(apply.LogAttrs(), "operation", operation, "fresh_state", fresh.State)...) + return false + } + + held, err := o.stor.ApplyComments().ClaimProgressCommentAuthority(ctx, o.applyID, o.authorityOwner) + if err != nil { + o.logger.Error("observer: failed to claim progress-comment authority; skipping GitHub side effect", + append(apply.LogAttrs(), "operation", operation, "authority_owner", o.authorityOwner, "error", err)...) + return false + } + if !held { + // The claim was not won: either another observer holds a fresh + // authority (its edits carry the PR), or no tracked progress comment + // row exists yet to claim (nothing to edit either way). Expected on + // every peer pod polling the same apply, hence Debug. + o.logger.Debug("observer: progress-comment authority not won (held by another observer, or no tracked comment row yet); skipping GitHub side effect", + append(apply.LogAttrs(), "operation", operation, "authority_owner", o.authorityOwner)...) + return false + } + return true +} + +// operationScopedWorkInFlight reports whether the apply's schema-change work +// is still in flight under operation-scoped dispatch — the one shape whose +// parent apply lease is legitimately unheld mid-apply. True when the apply is +// non-terminal and either its generation manifest still lists keys with no +// attached operation (the dispatcher owes more dispatch waves) or the apply +// has multiple attached operations with a keyed row not yet terminal. A +// single attached operation is deliberately not counted, matching the +// operator's drive-mode split: that shape drives under the parent apply +// lease, so an unheld lease there means no driver and the lease checks stay +// authoritative. (The one single-operation drive that runs under the +// operation lease — a task-less operation — fails closed here too; it has no +// task progress to report and its terminal comment is published by the +// aggregate projection winner.) Whole-deployment operations (empty key) are +// likewise not counted, whatever their number. +func (o *CommentObserver) operationScopedWorkInFlight(ctx context.Context, apply *storage.Apply) (bool, error) { + if state.IsTerminalApplyState(apply.State) { + return false, nil + } + ops, err := o.stor.ApplyOperations().ListByApply(ctx, o.applyID) + if err != nil { + return false, fmt.Errorf("load apply operations for progress-comment authority of apply %s: %w", apply.ApplyIdentifier, err) + } + if len(apply.MissingExpectedOperationKeys(ops)) > 0 { + return true, nil + } + if len(ops) <= 1 { + return false, nil + } + for _, op := range ops { + if op.OperationKey != "" && !state.IsApplyOperationTerminal(op.State) { + return true, nil + } + } + return false, nil +} + func (o *CommentObserver) contextWithApplyLease(ctx context.Context, apply *storage.Apply) context.Context { // The aggregate terminal observer holds the operation lease, not the parent // apply lease. Attaching an apply lease it does not hold would make every @@ -755,6 +948,13 @@ func (o *CommentObserver) contextWithApplyLease(ctx context.Context, apply *stor if o.aggregateTerminalCASWinner { return ctx } + // An apply row that records no parent lease has no lease to attach: the + // side-effect gate admitted this write under the progress-comment + // authority, so it takes storage's no-apply-lease path — the claim, not an + // apply lease, authorizes it, mirroring the aggregate CAS winner above. + if apply != nil && !apply.Lease().Valid() { + return ctx + } // Storage writes that record GitHub side effects must use the same lease as // the observer-side lease checks above. Attach the resolved lease even if it // is invalid so storage fails closed instead of performing an unleased write. diff --git a/pkg/webhook/comment_observer_test.go b/pkg/webhook/comment_observer_test.go index e595d36b7..2c63241ea 100644 --- a/pkg/webhook/comment_observer_test.go +++ b/pkg/webhook/comment_observer_test.go @@ -22,9 +22,11 @@ type stubApplyOperationStore struct { ops []*storage.ApplyOperation err error resumeByOp map[int64]*storage.EngineResumeState + listCalls int } func (s *stubApplyOperationStore) ListByApply(context.Context, int64) ([]*storage.ApplyOperation, error) { + s.listCalls++ return s.ops, s.err } @@ -40,11 +42,26 @@ func (s *stubApplyOperationStore) GetEngineResumeState(_ context.Context, opID i type stubStorage struct { storage.Storage ops storage.ApplyOperationStore + applies storage.ApplyStore settled []*storage.ApplyControlRequest } func (s *stubStorage) ApplyOperations() storage.ApplyOperationStore { return s.ops } +func (s *stubStorage) Applies() storage.ApplyStore { return s.applies } + +// stubApplyStore serves the authority gate's fresh re-read of the apply row +// from a fixed result. +type stubApplyStore struct { + storage.ApplyStore + apply *storage.Apply + err error +} + +func (s *stubApplyStore) Get(context.Context, int64) (*storage.Apply, error) { + return s.apply, s.err +} + func (s *stubStorage) Tasks() storage.TaskStore { return stubTaskStore{} } func (s *stubStorage) ControlRequests() storage.ControlRequestStore { @@ -224,10 +241,18 @@ type capturedLog struct { } type capturingLogger struct { + debugs []capturedLog + infos []capturedLog errors []capturedLog } -func (l *capturingLogger) Info(msg string, args ...any) {} +func (l *capturingLogger) Debug(msg string, args ...any) { + l.debugs = append(l.debugs, capturedLog{msg: msg, args: args}) +} + +func (l *capturingLogger) Info(msg string, args ...any) { + l.infos = append(l.infos, capturedLog{msg: msg, args: args}) +} func (l *capturingLogger) Error(msg string, args ...any) { l.errors = append(l.errors, capturedLog{msg: msg, args: args}) @@ -317,7 +342,7 @@ func TestAggregateTerminalObserverBypassesApplyLeaseCheck(t *testing.T) { normal := NewCommentObserver(cfg) assert.False(t, normal.leaseStillOwnsObserver(unleasedApply, "terminal"), - "a normal observer with no apply lease must fail closed") + "a terminal apply short-circuits the in-flight gate, so a normal observer with no apply lease fails closed") aggregate := NewAggregateTerminalCommentObserver(cfg) assert.True(t, aggregate.leaseStillOwnsObserver(unleasedApply, "terminal"), @@ -331,6 +356,184 @@ func TestAggregateTerminalObserverBypassesApplyLeaseCheck(t *testing.T) { assert.False(t, hasLease, "aggregate terminal observer must not attach an apply lease to storage writes") } +// The progress-comment authority exists only for applies whose work runs under +// operation leases while the parent apply lease is legitimately unheld. This +// pins the shape gate to the operator's drive-mode split: a multi-operation +// rollout with keyed work still in flight, or a generation manifest still +// promising operations, qualifies; terminal applies, settled keyed work, +// single-operation applies, and whole-deployment operations — whose drives +// hold the parent lease, so an unheld lease there means no driver — do not. +func TestOperationScopedWorkInFlight(t *testing.T) { + runningApply := &storage.Apply{ApplyIdentifier: "apply-1", State: state.Apply.Running} + + t.Run("terminal apply has no in-flight work", func(t *testing.T) { + // The operation rows alone would qualify — keyed, multiple, one still + // running — so only the apply's terminal state produces the denial. + o := newDispatchTestObserver(&stubApplyOperationStore{ops: []*storage.ApplyOperation{ + {ID: 1, Deployment: "primary", OperationKey: "orders/-80", State: state.ApplyOperation.Completed}, + {ID: 2, Deployment: "primary", OperationKey: "orders/80-", State: state.ApplyOperation.Running}, + }}) + inFlight, err := o.operationScopedWorkInFlight(t.Context(), &storage.Apply{ApplyIdentifier: "apply-1", State: state.Apply.Completed}) + require.NoError(t, err) + assert.False(t, inFlight) + }) + + t.Run("a single keyed operation drives under the parent lease", func(t *testing.T) { + o := newDispatchTestObserver(&stubApplyOperationStore{ops: []*storage.ApplyOperation{ + {ID: 1, Deployment: "primary", OperationKey: "orders/-80", State: state.ApplyOperation.Running}, + }}) + inFlight, err := o.operationScopedWorkInFlight(t.Context(), runningApply) + require.NoError(t, err) + assert.False(t, inFlight) + }) + + t.Run("non-terminal operation-keyed row is in flight", func(t *testing.T) { + o := newDispatchTestObserver(&stubApplyOperationStore{ops: []*storage.ApplyOperation{ + {ID: 1, Deployment: "primary", OperationKey: "orders/-80", State: state.ApplyOperation.Completed}, + {ID: 2, Deployment: "primary", OperationKey: "orders/80-", State: state.ApplyOperation.Running}, + }}) + inFlight, err := o.operationScopedWorkInFlight(t.Context(), runningApply) + require.NoError(t, err) + assert.True(t, inFlight) + }) + + t.Run("settled keyed work is not in flight", func(t *testing.T) { + o := newDispatchTestObserver(&stubApplyOperationStore{ops: []*storage.ApplyOperation{ + {ID: 1, Deployment: "primary", OperationKey: "orders/-80", State: state.ApplyOperation.Completed}, + {ID: 2, Deployment: "primary", OperationKey: "orders/80-", State: state.ApplyOperation.Completed}, + }}) + inFlight, err := o.operationScopedWorkInFlight(t.Context(), runningApply) + require.NoError(t, err) + assert.False(t, inFlight) + }) + + t.Run("whole-deployment operations are governed by the parent lease", func(t *testing.T) { + o := newDispatchTestObserver(&stubApplyOperationStore{ops: []*storage.ApplyOperation{ + {ID: 1, Deployment: "primary", State: state.ApplyOperation.Running}, + {ID: 2, Deployment: "replica", State: state.ApplyOperation.Running}, + }}) + inFlight, err := o.operationScopedWorkInFlight(t.Context(), runningApply) + require.NoError(t, err) + assert.False(t, inFlight) + }) + + t.Run("manifest keys with no attached operation are in flight", func(t *testing.T) { + manifestApply := &storage.Apply{ + ApplyIdentifier: "apply-1", + State: state.Apply.Running, + ExpectedOperationKeys: []string{"orders/-80", "orders/80-"}, + } + o := newDispatchTestObserver(&stubApplyOperationStore{ops: []*storage.ApplyOperation{ + {ID: 1, Deployment: "primary", OperationKey: "orders/-80", State: state.ApplyOperation.Completed}, + }}) + inFlight, err := o.operationScopedWorkInFlight(t.Context(), manifestApply) + require.NoError(t, err) + assert.True(t, inFlight) + }) + + t.Run("an operation-load failure is surfaced, never treated as in flight", func(t *testing.T) { + o := newDispatchTestObserver(&stubApplyOperationStore{err: errors.New("db unavailable")}) + inFlight, err := o.operationScopedWorkInFlight(t.Context(), runningApply) + require.Error(t, err) + assert.False(t, inFlight) + }) +} + +// inFlightKeyedOps is an operation set that qualifies for the durable +// progress-comment authority: a multi-operation rollout with keyed work still +// running. +func inFlightKeyedOps() []*storage.ApplyOperation { + return []*storage.ApplyOperation{ + {ID: 1, Deployment: "primary", OperationKey: "orders/-80", State: state.ApplyOperation.Completed}, + {ID: 2, Deployment: "primary", OperationKey: "orders/80-", State: state.ApplyOperation.Running}, + } +} + +// The observer's poller snapshot can be a tick old, so the authority gate +// re-reads the apply row before claiming. A dispatch wave that re-claimed the +// parent lease since the snapshot — or a projection that settled the apply +// terminal — owns the comment through its own path, and the authority must +// deny so two writers never edit the same comment concurrently. +func TestProgressCommentAuthorityDeniesOnFreshApplyRowChanges(t *testing.T) { + snapshot := &storage.Apply{ID: 7, ApplyIdentifier: "apply-1", State: state.Apply.Running} + + t.Run("parent lease re-claimed since the snapshot", func(t *testing.T) { + logger := &capturingLogger{} + o := newDispatchTestObserver(&stubApplyOperationStore{ops: inFlightKeyedOps()}) + o.logger = logger + o.stor.(*stubStorage).applies = &stubApplyStore{apply: &storage.Apply{ + ID: 7, ApplyIdentifier: "apply-1", State: state.Apply.Running, + LeaseOwner: "driver-host/9/dispatch", LeaseToken: "wave-token", + }} + + assert.False(t, o.progressCommentAuthorityOwnsObserver(snapshot, "progress"), + "a freshly claimed parent lease must deny the durable authority") + require.Len(t, logger.debugs, 1) + assert.Contains(t, logger.debugs[0].msg, "parent apply lease re-claimed") + }) + + t.Run("apply settled terminal since the snapshot", func(t *testing.T) { + logger := &capturingLogger{} + o := newDispatchTestObserver(&stubApplyOperationStore{ops: inFlightKeyedOps()}) + o.logger = logger + o.stor.(*stubStorage).applies = &stubApplyStore{apply: &storage.Apply{ + ID: 7, ApplyIdentifier: "apply-1", State: state.Apply.Completed, + }} + + assert.False(t, o.progressCommentAuthorityOwnsObserver(snapshot, "progress"), + "a freshly settled terminal apply must deny the durable authority") + require.Len(t, logger.debugs, 1) + assert.Contains(t, logger.debugs[0].msg, "apply settled terminal") + }) + + t.Run("a re-read failure fails closed", func(t *testing.T) { + logger := &capturingLogger{} + o := newDispatchTestObserver(&stubApplyOperationStore{ops: inFlightKeyedOps()}) + o.logger = logger + o.stor.(*stubStorage).applies = &stubApplyStore{err: errors.New("db unavailable")} + + assert.False(t, o.progressCommentAuthorityOwnsObserver(snapshot, "progress")) + require.Len(t, logger.errors, 1) + assert.Contains(t, logger.errors[0].msg, "failed to re-read the apply") + }) + + t.Run("a vanished apply row fails closed", func(t *testing.T) { + logger := &capturingLogger{} + o := newDispatchTestObserver(&stubApplyOperationStore{ops: inFlightKeyedOps()}) + o.logger = logger + o.stor.(*stubStorage).applies = &stubApplyStore{} + + assert.False(t, o.progressCommentAuthorityOwnsObserver(snapshot, "progress")) + require.Len(t, logger.errors, 1) + assert.Contains(t, logger.errors[0].msg, "apply row no longer exists") + }) +} + +// One observer callback checks the side-effect gate several times — before the +// comment lookup, the client creation, the edit, and the tracking write. The +// authority decision is made once per callback against fresh storage rows and +// then reused, so a tick issues one operation scan and one claim write rather +// than one per gate check. Resetting the decision (the next callback) decides +// afresh. +func TestProgressCommentAuthorityDecidesOncePerCallback(t *testing.T) { + snapshot := &storage.Apply{ID: 7, ApplyIdentifier: "apply-1", State: state.Apply.Running} + opStore := &stubApplyOperationStore{ops: inFlightKeyedOps()} + o := newDispatchTestObserver(opStore) + o.logger = &capturingLogger{} + o.stor.(*stubStorage).applies = &stubApplyStore{apply: &storage.Apply{ + ID: 7, ApplyIdentifier: "apply-1", State: state.Apply.Running, + LeaseOwner: "driver-host/9/dispatch", LeaseToken: "wave-token", + }} + + assert.False(t, o.progressCommentAuthorityOwnsObserver(snapshot, "progress")) + assert.False(t, o.progressCommentAuthorityOwnsObserver(snapshot, "edit GitHub comment")) + assert.Equal(t, 1, opStore.listCalls, "the second gate check in one callback must reuse the decision") + + o.resetProgressCommentAuthorityDecision() + assert.False(t, o.progressCommentAuthorityOwnsObserver(snapshot, "progress")) + assert.Equal(t, 2, opStore.listCalls, "the next callback must decide afresh") +} + // A per-driver observer for a multi-operation apply must not publish a separate // apply-level summary comment: it holds only one operation's task slice, so // publishing here would post a duplicate, partial summary. The aggregate