fix(github): keep the PR progress comment updating between operation dispatch waves

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,

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,

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(&current)
+	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

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) {

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

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

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))