From adca4ca21d7a7f024a1c8b164e38915f1a88da5b Mon Sep 17 00:00:00 2001 From: Tien Dung Dao Date: Sat, 29 Aug 2026 12:46:02 +0700 Subject: [PATCH 1/2] fix(mcp): resolve superseded failed mutation receipts on later success MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A mutation whose disk write succeeded but whose graph ingest failed leaves a terminally failed receipt in Server.mutationReceipts. The freshness barrier then refuses change.detect/change.impact for the whole repository until the receipt's 10-minute retention lapses, even though waiting can never heal a terminal failure — and a later generation of the same path that ingests successfully already proves the graph reflects newer bytes than the failed generation ever wrote. Drop terminally failed receipts for a path when a later generation of that path completes successfully. Pending receipts and failures at or above the succeeded generation are untouched. Also state in the barrier error that terminally failed generations do not recover by waiting. Fixes the rolling repo-wide detect lockout described in #692: under load, each failed ingest of an actively edited file re-armed the barrier for another retention window, which agents could neither clear nor reconcile. Co-Authored-By: Claude Fable 5 --- internal/mcp/edit_serialization.go | 38 ++++++++++++ internal/mcp/mutation_freshness_test.go | 82 +++++++++++++++++++++++++ 2 files changed, 120 insertions(+) diff --git a/internal/mcp/edit_serialization.go b/internal/mcp/edit_serialization.go index c6664ccea..0af1d7ea4 100644 --- a/internal/mcp/edit_serialization.go +++ b/internal/mcp/edit_serialization.go @@ -171,6 +171,9 @@ func (s *Server) trackMutationTicket(ticket *indexer.MutationTicket) *mutationRe receipt.result = result receipt.completed = true receipt.mu.Unlock() + if result.Err == nil && result.Reindexed { + s.resolveSupersededFailedReceipts(receipt) + } close(receipt.done) time.AfterFunc(mutationReceiptRetention, func() { s.mutationReceipts.Delete(receipt.id) @@ -179,6 +182,33 @@ func (s *Server) trackMutationTicket(ticket *indexer.MutationTicket) *mutationRe return receipt } +// resolveSupersededFailedReceipts drops terminally failed receipts for a path +// once a later generation of the same path has been applied successfully. The +// graph then reflects newer bytes than the failed generation ever wrote, so +// the stale failure no longer describes a real freshness gap — keeping it +// would only fail freshness barriers that waiting cannot heal, because a +// terminal error never completes differently. Pending receipts and failures +// at or above the succeeded generation are left untouched. +func (s *Server) resolveSupersededFailedReceipts(succeeded *mutationReceipt) { + succeededPath := filepath.Clean(succeeded.path) + s.mutationReceipts.Range(func(key, value any) bool { + other, ok := value.(*mutationReceipt) + if !ok || other == succeeded { + return true + } + if other.generation >= succeeded.generation || filepath.Clean(other.path) != succeededPath { + return true + } + other.mu.RLock() + terminalFailure := other.completed && (other.result.Err != nil || !other.result.Reindexed) + other.mu.RUnlock() + if terminalFailure { + s.mutationReceipts.Delete(key) + } + return true + }) +} + func (r *mutationReceipt) outcome(pending bool) mutationReindexOutcome { r.mu.RLock() defer r.mu.RUnlock() @@ -313,16 +343,19 @@ waitLoop: } issues := make([]string, 0, len(receipts)) + hasTerminalFailure := false for _, receipt := range receipts { select { case <-receipt.done: outcome := receipt.outcome(false) switch { case outcome.Err != nil: + hasTerminalFailure = true issues = append(issues, fmt.Sprintf( "failed receipt=%s repo=%q path=%q generation=%d error=%q", receipt.id, receipt.repo, receipt.path, receipt.generation, outcome.Err.Error())) case !outcome.Reindexed: + hasTerminalFailure = true issues = append(issues, fmt.Sprintf( "failed receipt=%s repo=%q path=%q generation=%d error=%q", receipt.id, receipt.repo, receipt.path, receipt.generation, "reindex not confirmed")) @@ -352,6 +385,11 @@ waitLoop: } message += issue } + if hasTerminalFailure { + message += "; terminally failed generations do not recover by waiting — " + + "they clear when a later mutation of the same path succeeds or when " + + "the receipt retention lapses" + } return fmt.Errorf("%s", message) } diff --git a/internal/mcp/mutation_freshness_test.go b/internal/mcp/mutation_freshness_test.go index e632fe555..a921001cc 100644 --- a/internal/mcp/mutation_freshness_test.go +++ b/internal/mcp/mutation_freshness_test.go @@ -104,6 +104,88 @@ func TestMutationFreshnessTerminalFailureFailsClosed(t *testing.T) { } } +func TestMutationFreshnessSuccessResolvesSupersededFailures(t *testing.T) { + s := &Server{mutationSafetyWait: time.Millisecond} + stale := pendingFreshnessReceipt(s, "receipt-stale-failed", "repo-a", "/repo-a/file.go", 6) + completeFreshnessReceipt(stale, indexer.MutationResult{ + RequestedGeneration: 6, + Err: errors.New("context deadline exceeded"), + }) + otherPath := pendingFreshnessReceipt(s, "receipt-other-path", "repo-a", "/repo-a/other.go", 7) + completeFreshnessReceipt(otherPath, indexer.MutationResult{ + RequestedGeneration: 7, + Err: errors.New("unrelated failure"), + }) + newer := pendingFreshnessReceipt(s, "receipt-newer-failed", "repo-a", "/repo-a/file.go", 12) + completeFreshnessReceipt(newer, indexer.MutationResult{ + RequestedGeneration: 12, + Err: errors.New("later failure"), + }) + + succeeded := pendingFreshnessReceipt(s, "receipt-success", "repo-a", "/repo-a/file.go", 9) + completeFreshnessReceipt(succeeded, indexer.MutationResult{ + RequestedGeneration: 9, + AppliedGeneration: 9, + Reindexed: true, + }) + s.resolveSupersededFailedReceipts(succeeded) + + if _, loaded := s.mutationReceipts.Load("receipt-stale-failed"); loaded { + t.Fatal("superseded failed receipt survived a later successful generation") + } + if _, loaded := s.mutationReceipts.Load("receipt-other-path"); !loaded { + t.Fatal("failure on an unrelated path was dropped") + } + if _, loaded := s.mutationReceipts.Load("receipt-newer-failed"); !loaded { + t.Fatal("failure newer than the succeeded generation was dropped") + } + + err := s.awaitMutationFreshnessForRepos(context.Background(), "repo-a") + if err == nil { + t.Fatal("remaining failures did not fail closed") + } + message := err.Error() + if strings.Contains(message, "receipt-stale-failed") { + t.Fatalf("freshness error still reports the superseded receipt: %s", message) + } + for _, want := range []string{ + "receipt-other-path", + "receipt-newer-failed", + "do not recover by waiting", + } { + if !strings.Contains(message, want) { + t.Fatalf("freshness error %q does not contain %q", message, want) + } + } +} + +func TestTrackMutationTicketResolvesSupersededFailures(t *testing.T) { + s := &Server{} + stale := pendingFreshnessReceipt(s, "receipt-stale-failed", "", "/repo/file.go", 3) + completeFreshnessReceipt(stale, indexer.MutationResult{ + RequestedGeneration: 3, + Err: errors.New("context deadline exceeded"), + }) + + done := make(chan indexer.MutationResult, 1) + ticket := &indexer.MutationTicket{Path: "/repo/file.go", Generation: 5, Done: done} + receipt := s.trackMutationTicket(ticket) + done <- indexer.MutationResult{ + RequestedGeneration: 5, + AppliedGeneration: 5, + Reindexed: true, + } + close(done) + <-receipt.done + + if _, loaded := s.mutationReceipts.Load("receipt-stale-failed"); loaded { + t.Fatal("stale failed receipt not resolved after a successful ticket") + } + if _, loaded := s.mutationReceipts.Load(receipt.id); !loaded { + t.Fatal("successful receipt itself was dropped before retention") + } +} + func TestMutationReposForSymbolIDsUnresolvedWidensBarrier(t *testing.T) { g := graph.New() g.AddNode(&graph.Node{ From 027959d6975d7aa0d80bbdd384b2299daafb4c3a Mon Sep 17 00:00:00 2001 From: Tien Dung Dao Date: Sat, 29 Aug 2026 14:54:54 +0700 Subject: [PATCH 2/2] fix(mcp): resolve superseded failures in place instead of deleting Review follow-up: deleting a superseded receipt left the mutation-commit ledger's graph half refreshing against a missing id, so a record whose edit response had timed out reported graph_status "pending" forever, with guidance telling the caller to keep waiting. Stamp the superseding apply onto the failed receipt instead, mirroring how completeMutationWaiters resolves earlier waiters with the later apply's result; the receipt stays queryable and reports fresh with the superseding applied generation. The succeeded result is passed by value so the sweep holds no lock other than the receipt it is stamping, removing the cross-receipt lock shape. Add the missing pending-receipt guard test: a still-in-flight receipt for the same path must survive the resolve untouched and still fail the freshness barrier. All four guards now bind under mutation (call removed, generation guard dropped, path guard dropped, pending stamped): each mutant fails its test and the restored suite passes. Co-Authored-By: Claude Fable 5 --- internal/mcp/edit_serialization.go | 47 +++++++++++++-------- internal/mcp/mutation_freshness_test.go | 56 ++++++++++++++++++++----- 2 files changed, 74 insertions(+), 29 deletions(-) diff --git a/internal/mcp/edit_serialization.go b/internal/mcp/edit_serialization.go index 0af1d7ea4..089d73230 100644 --- a/internal/mcp/edit_serialization.go +++ b/internal/mcp/edit_serialization.go @@ -172,7 +172,7 @@ func (s *Server) trackMutationTicket(ticket *indexer.MutationTicket) *mutationRe receipt.completed = true receipt.mu.Unlock() if result.Err == nil && result.Reindexed { - s.resolveSupersededFailedReceipts(receipt) + s.resolveSupersededFailedReceipts(receipt.path, receipt.generation, result) } close(receipt.done) time.AfterFunc(mutationReceiptRetention, func() { @@ -182,29 +182,40 @@ func (s *Server) trackMutationTicket(ticket *indexer.MutationTicket) *mutationRe return receipt } -// resolveSupersededFailedReceipts drops terminally failed receipts for a path -// once a later generation of the same path has been applied successfully. The -// graph then reflects newer bytes than the failed generation ever wrote, so -// the stale failure no longer describes a real freshness gap — keeping it -// would only fail freshness barriers that waiting cannot heal, because a -// terminal error never completes differently. Pending receipts and failures -// at or above the succeeded generation are left untouched. -func (s *Server) resolveSupersededFailedReceipts(succeeded *mutationReceipt) { - succeededPath := filepath.Clean(succeeded.path) - s.mutationReceipts.Range(func(key, value any) bool { +// resolveSupersededFailedReceipts resolves terminally failed receipts for a +// path once a later generation of the same path has been applied +// successfully. The graph then reflects newer bytes than the failed +// generation ever wrote, so the stale failure no longer describes a real +// freshness gap — keeping it would only fail freshness barriers that waiting +// cannot heal, because a terminal error never completes differently. +// +// The failed receipt is resolved in place rather than deleted: the +// mutation-commit ledger refreshes its graph half through +// mutationReceiptState, and a deleted receipt would leave that record +// reading "pending" forever. Stamping the superseding apply mirrors how +// completeMutationWaiters resolves earlier waiters with the later apply's +// result. Pending receipts and failures at or above the succeeded +// generation are left untouched. The succeeded result is passed by value so +// the sweep holds no lock besides the receipt it is stamping. +func (s *Server) resolveSupersededFailedReceipts(succeededPath string, succeededGeneration uint64, applied indexer.MutationResult) { + cleanPath := filepath.Clean(succeededPath) + s.mutationReceipts.Range(func(_, value any) bool { other, ok := value.(*mutationReceipt) - if !ok || other == succeeded { + if !ok { return true } - if other.generation >= succeeded.generation || filepath.Clean(other.path) != succeededPath { + if other.generation >= succeededGeneration || filepath.Clean(other.path) != cleanPath { return true } - other.mu.RLock() - terminalFailure := other.completed && (other.result.Err != nil || !other.result.Reindexed) - other.mu.RUnlock() - if terminalFailure { - s.mutationReceipts.Delete(key) + other.mu.Lock() + if other.completed && (other.result.Err != nil || !other.result.Reindexed) { + other.result = indexer.MutationResult{ + RequestedGeneration: other.generation, + AppliedGeneration: applied.AppliedGeneration, + Reindexed: true, + } } + other.mu.Unlock() return true }) } diff --git a/internal/mcp/mutation_freshness_test.go b/internal/mcp/mutation_freshness_test.go index a921001cc..397cf85ff 100644 --- a/internal/mcp/mutation_freshness_test.go +++ b/internal/mcp/mutation_freshness_test.go @@ -123,21 +123,26 @@ func TestMutationFreshnessSuccessResolvesSupersededFailures(t *testing.T) { }) succeeded := pendingFreshnessReceipt(s, "receipt-success", "repo-a", "/repo-a/file.go", 9) - completeFreshnessReceipt(succeeded, indexer.MutationResult{ + appliedResult := indexer.MutationResult{ RequestedGeneration: 9, AppliedGeneration: 9, Reindexed: true, - }) - s.resolveSupersededFailedReceipts(succeeded) + } + completeFreshnessReceipt(succeeded, appliedResult) + s.resolveSupersededFailedReceipts("/repo-a/file.go", 9, appliedResult) - if _, loaded := s.mutationReceipts.Load("receipt-stale-failed"); loaded { - t.Fatal("superseded failed receipt survived a later successful generation") + if _, loaded := s.mutationReceipts.Load("receipt-stale-failed"); !loaded { + t.Fatal("superseded receipt was deleted; it must stay queryable for the mutation-commit ledger") + } + resolved := stale.outcome(false) + if resolved.Err != nil || !resolved.Reindexed || resolved.AppliedGeneration != 9 { + t.Fatalf("superseded failure was not resolved in place: %+v", resolved) } - if _, loaded := s.mutationReceipts.Load("receipt-other-path"); !loaded { - t.Fatal("failure on an unrelated path was dropped") + if outcome := otherPath.outcome(false); outcome.Err == nil { + t.Fatal("failure on an unrelated path was resolved") } - if _, loaded := s.mutationReceipts.Load("receipt-newer-failed"); !loaded { - t.Fatal("failure newer than the succeeded generation was dropped") + if outcome := newer.outcome(false); outcome.Err == nil { + t.Fatal("failure newer than the succeeded generation was resolved") } err := s.awaitMutationFreshnessForRepos(context.Background(), "repo-a") @@ -178,14 +183,43 @@ func TestTrackMutationTicketResolvesSupersededFailures(t *testing.T) { close(done) <-receipt.done - if _, loaded := s.mutationReceipts.Load("receipt-stale-failed"); loaded { - t.Fatal("stale failed receipt not resolved after a successful ticket") + if _, loaded := s.mutationReceipts.Load("receipt-stale-failed"); !loaded { + t.Fatal("stale receipt was deleted; it must stay queryable for the mutation-commit ledger") + } + resolved := stale.outcome(false) + if resolved.Err != nil || !resolved.Reindexed || resolved.AppliedGeneration != 5 { + t.Fatalf("stale failed receipt not resolved after a successful ticket: %+v", resolved) } if _, loaded := s.mutationReceipts.Load(receipt.id); !loaded { t.Fatal("successful receipt itself was dropped before retention") } } +func TestMutationFreshnessSuccessKeepsPendingSamePathReceipts(t *testing.T) { + s := &Server{mutationSafetyWait: time.Millisecond} + inflight := pendingFreshnessReceipt(s, "receipt-inflight", "repo-a", "/repo-a/file.go", 4) + + appliedResult := indexer.MutationResult{ + RequestedGeneration: 9, + AppliedGeneration: 9, + Reindexed: true, + } + succeeded := pendingFreshnessReceipt(s, "receipt-success", "repo-a", "/repo-a/file.go", 9) + completeFreshnessReceipt(succeeded, appliedResult) + s.resolveSupersededFailedReceipts("/repo-a/file.go", 9, appliedResult) + + if _, loaded := s.mutationReceipts.Load("receipt-inflight"); !loaded { + t.Fatal("a still-pending receipt for the same path was dropped by the resolve") + } + if outcome := inflight.outcome(true); outcome.Pending != true { + t.Fatalf("a still-pending receipt was marked completed by the resolve: %+v", outcome) + } + err := s.awaitMutationFreshnessForRepos(context.Background(), "repo-a") + if err == nil || !strings.Contains(err.Error(), "receipt-inflight") { + t.Fatalf("barrier no longer reports the in-flight receipt: %v", err) + } +} + func TestMutationReposForSymbolIDsUnresolvedWidensBarrier(t *testing.T) { g := graph.New() g.AddNode(&graph.Node{