From 77f4fdd7fbf735ddd3c25f226f08a544c6c76818 Mon Sep 17 00:00:00 2001 From: Chandan Kumar Date: Sun, 27 Sep 2026 17:56:38 +0530 Subject: [PATCH] fix(vault): retry body indexing after a failed attempt The enrich worker recorded a doc in its dedup map even when writing its body chunks failed, so the next event with the same content hash was skipped and the doc stayed without chunks until a manual rescan. indexBody now returns its error and processChunk leaves those docs out of the dedup map, so the next event for the same hash tries again. Docs whose chunks were written are still deduplicated as before. --- internal/vault/body_index.go | 9 +- internal/vault/enrich_body_retry_test.go | 107 +++++++++++++++++++++++ internal/vault/enrich_worker.go | 12 ++- 3 files changed, 123 insertions(+), 5 deletions(-) create mode 100644 internal/vault/enrich_body_retry_test.go diff --git a/internal/vault/body_index.go b/internal/vault/body_index.go index 12ccbf7a..64cb264a 100644 --- a/internal/vault/body_index.go +++ b/internal/vault/body_index.go @@ -38,15 +38,18 @@ func chunkBody(raw []byte) []store.VaultChunk { // indexBody rebuilds the body chunks of one text document from its workspace // file. The store skips the write when the chunks already match contentHash. -func indexBody(ctx context.Context, vs store.VaultStore, tenantID, docID, contentHash, fullPath string) { +// Failures are logged here; the error only tells the caller to retry later. +func indexBody(ctx context.Context, vs store.VaultStore, tenantID, docID, contentHash, fullPath string) error { raw, err := os.ReadFile(fullPath) if err != nil { slog.Warn("vault.body_index: read_file", "path", fullPath, "err", err) - return + return err } if err := vs.ReplaceDocumentChunks(ctx, tenantID, docID, contentHash, chunkBody(raw)); err != nil { slog.Warn("vault.body_index: replace_chunks", "doc", docID, "err", err) + return err } + return nil } // bodyIndexRuns keeps one IndexStaleBodies pass per tenant. @@ -72,7 +75,7 @@ func IndexStaleBodies(ctx context.Context, vs store.VaultStore, tenantID, worksp if ctx.Err() != nil { return true } - indexBody(ctx, vs, tenantID, doc.ID, doc.ContentHash, filepath.Join(workspace, doc.Path)) + _ = indexBody(ctx, vs, tenantID, doc.ID, doc.ContentHash, filepath.Join(workspace, doc.Path)) } if len(docs) > 0 { slog.Info("vault.body_index: backfilled", "tenant", tenantID, "count", len(docs)) diff --git a/internal/vault/enrich_body_retry_test.go b/internal/vault/enrich_body_retry_test.go new file mode 100644 index 00000000..a53b26f4 --- /dev/null +++ b/internal/vault/enrich_body_retry_test.go @@ -0,0 +1,107 @@ +package vault + +import ( + "context" + "errors" + "os" + "path/filepath" + "sync" + "testing" + + "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/eventbus" + "github.com/nextlevelbuilder/goclaw/internal/providers" + "github.com/nextlevelbuilder/goclaw/internal/store" + "golang.org/x/sync/semaphore" +) + +// fakeVaultStoreBodyRetry fails ReplaceDocumentChunks a set number of times, +// then stores the chunks. Everything else the enrich pipeline touches is a no-op. +type fakeVaultStoreBodyRetry struct { + store.VaultStore + doc store.VaultDocument + failsLeft int + calls int + chunks []store.VaultChunk +} + +func (f *fakeVaultStoreBodyRetry) GetDocumentsByIDs(ctx context.Context, tenantID string, ids []string) ([]store.VaultDocument, error) { + return []store.VaultDocument{f.doc}, nil +} + +func (f *fakeVaultStoreBodyRetry) ReplaceDocumentChunks(ctx context.Context, tenantID, docID, contentHash string, chunks []store.VaultChunk) error { + f.calls++ + if f.failsLeft > 0 { + f.failsLeft-- + return errors.New("connection reset") + } + f.chunks = chunks + return nil +} + +func (f *fakeVaultStoreBodyRetry) UpdateSummaryAndReembed(ctx context.Context, tenantID, docID, summary string) error { + return nil +} + +func (f *fakeVaultStoreBodyRetry) FindSimilarDocs(ctx context.Context, tenantID, agentID, docID string, limit int) ([]store.VaultSearchResult, error) { + return nil, nil +} + +func (f *fakeVaultStoreBodyRetry) GetDocumentByID(ctx context.Context, tenantID, id string) (*store.VaultDocument, error) { + return nil, errors.New("not needed") +} + +// A transient body-index failure must not mark the doc as processed, or the +// next event with the same hash is dropped and the doc never gets chunks. +func TestEnrichWorker_RetriesBodyIndexAfterFailure(t *testing.T) { + ws := t.TempDir() + if err := os.WriteFile(filepath.Join(ws, "prices.md"), []byte("Deluxe Ocean Suite 4,200,000\n"), 0o644); err != nil { + t.Fatal(err) + } + tenantID := providers.MasterTenantID.String() + docID := uuid.NewString() + fs := &fakeVaultStoreBodyRetry{ + doc: store.VaultDocument{ + ID: docID, TenantID: tenantID, Path: "prices.md", DocType: "note", + ContentHash: "h1", Summary: "Hotel room rates.", + }, + failsLeft: 1, + } + reg := providers.NewRegistry(nil) + reg.Register(&mockClassifyProvider{}) + w := &EnrichWorker{ + vault: fs, + registry: reg, + dedup: make(map[string]string), + sem: semaphore.NewWeighted(enrichMaxConcurrent), + progress: NewEnrichProgress(nil), + cancelFuncs: &sync.Map{}, + } + event := eventbus.DomainEvent{Payload: eventbus.VaultDocUpsertedPayload{ + DocID: docID, TenantID: tenantID, Path: "prices.md", ContentHash: "h1", Workspace: ws, + }} + + ctx := context.Background() + if err := w.Handle(ctx, event); err != nil { + t.Fatalf("first Handle: %v", err) + } + if fs.chunks != nil { + t.Fatal("chunks written although ReplaceDocumentChunks failed") + } + + // Same event again, e.g. the next file save or sync tick. + if err := w.Handle(ctx, event); err != nil { + t.Fatalf("second Handle: %v", err) + } + if len(fs.chunks) != 1 || fs.chunks[0].Text != "Deluxe Ocean Suite 4,200,000" { + t.Fatalf("chunks after replay = %+v, want the file body", fs.chunks) + } + + // Once indexed, the dedup gate applies again. + if err := w.Handle(ctx, event); err != nil { + t.Fatalf("third Handle: %v", err) + } + if fs.calls != 2 { + t.Fatalf("ReplaceDocumentChunks calls = %d, want 2", fs.calls) + } +} diff --git a/internal/vault/enrich_worker.go b/internal/vault/enrich_worker.go index 699128e3..af225ae7 100644 --- a/internal/vault/enrich_worker.go +++ b/internal/vault/enrich_worker.go @@ -324,9 +324,15 @@ func (w *EnrichWorker) processChunk(ctx context.Context, items []eventbus.VaultD } // Body chunks need no LLM, so they are indexed even without a provider. + // Docs whose body failed to index are kept out of the dedup map below, + // otherwise the next event for the same hash would be skipped and the + // doc would stay without chunks until someone runs a rescan. + bodyFailed := make(map[string]bool) for _, item := range pending { if doc := docMap[item.DocID]; doc != nil && doc.DocType != "media" && doc.DocType != "document" { - indexBody(ctx, w.vault, tenantID, item.DocID, item.ContentHash, filepath.Join(item.Workspace, item.Path)) + if err := indexBody(ctx, w.vault, tenantID, item.DocID, item.ContentHash, filepath.Join(item.Workspace, item.Path)); err != nil { + bodyFailed[item.DocID] = true + } } } @@ -454,7 +460,9 @@ func (w *EnrichWorker) processChunk(ctx context.Context, items []eventbus.VaultD // Phase 4 — Record dedup + wikilinks. for _, r := range embedded { - w.recordDedup(r.payload.DocID, r.payload.ContentHash) + if !bodyFailed[r.payload.DocID] { + w.recordDedup(r.payload.DocID, r.payload.ContentHash) + } w.syncWikilinks(ctx, r.payload) } }