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