From 97ec9e7ce8420437281da19189e06ea66fd55888 Mon Sep 17 00:00:00 2001 From: Mathias Date: Mon, 19 Oct 2026 21:06:46 +0200 Subject: [PATCH] fix(watcher): keep a note that changed while it was being ingested The archive holds the content read at the start; if Syncthing replaced the note during the (minutes-long) LLM pass, removing the original would drop the newer content. Compare size+mtime before removing; leave it for the next poll. Co-Authored-By: Claude Sonnet 5.5 --- ingestion/internal/watcher/watcher.go | 23 +++++++++++++++++++++ ingestion/internal/watcher/watcher_test.go | 24 ++++++++++++++++++++++ 2 files changed, 47 insertions(+) diff --git a/ingestion/internal/watcher/watcher.go b/ingestion/internal/watcher/watcher.go index 5feb3fd..f46bf07 100644 --- a/ingestion/internal/watcher/watcher.go +++ b/ingestion/internal/watcher/watcher.go @@ -156,6 +156,10 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d return fmt.Errorf("relative path: %w", err) } + before, err := root.Lstat(rel) + if err != nil { + return fmt.Errorf("stat before ingest: %w", err) + } content, err := extract.TextIn(root, rel) if err != nil { return fmt.Errorf("extract text: %w", err) @@ -172,6 +176,10 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d if cpErr := copyFromRoot(root, rel, dest); cpErr != nil { return fmt.Errorf("copy to failed: %w", cpErr) } + if changedSince(root, rel, before) { + slog.Warn("watcher: note changed during ingest; leaving it in raw/ for the next poll", "file", filename) + return nil + } retire(cfg, root, rel, path, ".failed", runErr.Error()) slog.Warn("watcher: file failed", "file", filename, "error", runErr) @@ -198,12 +206,27 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d if err := copyFromRoot(root, rel, dest); err != nil { return fmt.Errorf("copy to processed: %w", err) } + if changedSince(root, rel, before) { + slog.Warn("watcher: note changed during ingest; leaving it in raw/ for the next poll", "file", filename) + return nil + } retire(cfg, root, rel, path, ".processed", date) slog.Info("watcher: file processed", "file", filename, "source", source) return nil } +// changedSince reports whether the note at rel differs (size or mtime) from +// before. A note replaced mid-ingest (Syncthing) has newer content than the +// archive copy, so it must not be removed. +func changedSince(root *os.Root, rel string, before os.FileInfo) bool { + now, err := root.Lstat(rel) + if err != nil { + return true + } + return now.Size() != before.Size() || !now.ModTime().Equal(before.ModTime()) +} + // retire removes an archived original from raw/ (inbox zero). If the removal // fails it falls back to a marker file so the watcher still skips the file. func retire(cfg Config, root *os.Root, rel, path, markerSuffix, markerBody string) { diff --git a/ingestion/internal/watcher/watcher_test.go b/ingestion/internal/watcher/watcher_test.go index 9af0e7d..4b0f547 100644 --- a/ingestion/internal/watcher/watcher_test.go +++ b/ingestion/internal/watcher/watcher_test.go @@ -444,3 +444,27 @@ func TestProcessDir_KeepsTheOriginalWhenTheArchiveCopyFails(t *testing.T) { _, err := os.Stat(note) assert.NoError(t, err, "original stays in raw/ when it could not be archived") } + +// If the note is replaced (e.g. by Syncthing) while it is being ingested, the +// archive holds the OLD content; removing the original would lose the new one. +func TestProcessDir_KeepsANoteThatChangedWhileItWasBeingIngested(t *testing.T) { + brainDir := setupBrainDir(t) + note := filepath.Join(brainDir, "raw", "edited.md") + require.NoError(t, os.WriteFile(note, []byte("old"), 0o644)) + + cfg := Config{ + BrainDir: brainDir, + Pipeline: pipeline.Config{ + Complete: func(ctx context.Context, s, u string) (string, error) { + require.NoError(t, os.WriteFile(note, []byte("newer and longer content"), 0o644)) + return successComplete(pipeline.RawPage{Title: "Edited", Type: "source", Subtype: "article", Content: "x\n"})(ctx, s, u) + }, + Schema: "# Schema", + }, + } + processDir(context.Background(), cfg, "2026-10-19") + + got, err := os.ReadFile(note) + require.NoError(t, err, "a note that changed during ingest must stay in raw/") + assert.Equal(t, "newer and longer content", string(got)) +}