diff --git a/ingestion/internal/watcher/watcher.go b/ingestion/internal/watcher/watcher.go index b5c41b3..2a951a3 100644 --- a/ingestion/internal/watcher/watcher.go +++ b/ingestion/internal/watcher/watcher.go @@ -9,6 +9,7 @@ import ( "os" "path/filepath" "strings" + "sync" "time" "unicode" @@ -21,11 +22,19 @@ type Config struct { BrainDir string Interval time.Duration Pipeline pipeline.Config + + // stuck remembers notes that were ingested but could neither be removed + // from raw/ nor marked (e.g. a read-only raw/), so later polls skip them + // instead of ingesting them again. In memory only: a restart retries once. + stuck *sync.Map } // Start launches the watcher in a background goroutine. // It returns immediately. The watcher stops when ctx is cancelled. func Start(ctx context.Context, cfg Config) { + if cfg.stuck == nil { + cfg.stuck = &sync.Map{} + } go func() { ticker := time.NewTicker(cfg.Interval) defer ticker.Stop() @@ -108,6 +117,12 @@ func processDir(ctx context.Context, cfg Config, date string) []error { return nil } + if cfg.stuck != nil { + if _, skip := cfg.stuck.Load(path); skip { + return nil + } + } + // Skip files that have already been processed or permanently failed. if _, err := os.Stat(path + ".processed"); err == nil { return nil @@ -127,9 +142,12 @@ func processDir(ctx context.Context, cfg Config, date string) []error { return errs } -// processFile reads a file, calls pipeline.Run, copies it to processed/ or failed/, -// and writes a marker file next to the original so the watcher skips it next poll. -// The original file is never deleted, keeping Syncthing-connected vaults (e.g. Obsidian) intact. +// processFile reads a file, calls pipeline.Run, and archives it into processed/ +// or failed/. The original is then removed from raw/ so that raw/ only ever +// holds unprocessed input (inbox zero). The removal syncs to Syncthing peers; +// the archive copy under raw/processed/ or raw/failed/ is what they keep. A +// marker is written next to the original only if the removal fails, so a +// stuck file is not re-ingested every poll. func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, date string) error { filename := filepath.Base(path) source := deriveSource(filename) @@ -138,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) @@ -154,9 +176,11 @@ 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 mkErr := os.WriteFile(path+".failed", []byte(runErr.Error()), 0o644); mkErr != nil { - slog.Error("watcher: failed to write .failed marker", "error", mkErr) + 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) @@ -182,14 +206,45 @@ 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 err := os.WriteFile(path+".processed", []byte(date), 0o644); err != nil { - slog.Error("watcher: failed to write .processed marker", "error", 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) was extracted from older +// content than it now holds, so it must not be removed. Limits: a same-size +// rewrite with a restored mtime is not detected (the new bytes then survive +// only in the archive copy), nor a change between this check and Remove. +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) { + rmErr := root.Remove(rel) + if rmErr == nil { + return + } + slog.Error("watcher: failed to remove archived original; writing marker", "file", rel, "error", rmErr) + if err := os.WriteFile(path+markerSuffix, []byte(markerBody), 0o644); err != nil { + slog.Error("watcher: failed to write marker; skipping this note until restart", "file", rel, "error", err) + if cfg.stuck != nil { + cfg.stuck.Store(path, struct{}{}) + } + } +} + // deriveSource turns a filename into a human-readable source name. // "shape-up-book.md" → "Shape Up Book" func deriveSource(filename string) string { diff --git a/ingestion/internal/watcher/watcher_test.go b/ingestion/internal/watcher/watcher_test.go index 899b2bf..1f5670d 100644 --- a/ingestion/internal/watcher/watcher_test.go +++ b/ingestion/internal/watcher/watcher_test.go @@ -89,13 +89,12 @@ func TestStart_ProcessesFile(t *testing.T) { } require.True(t, found, "file should be copied to processed/") - // Original file should still exist (copy, not move — keeps Obsidian vault intact). + // Inbox zero: once archived, the original leaves raw/ and no marker is + // left behind, so what remains in raw/ is exactly the unprocessed set. _, err := os.Stat(rawFile) - assert.NoError(t, err, "original file should remain in raw/") - - // A .processed marker should exist next to the original. + assert.ErrorIs(t, err, os.ErrNotExist, "original should be removed from raw/ once archived") _, err = os.Stat(rawFile + ".processed") - assert.NoError(t, err, ".processed marker should be written") + assert.ErrorIs(t, err, os.ErrNotExist, "no .processed marker: the archive copy is the record") // Wiki page should exist. wikiPath := filepath.Join(brainDir, "wiki", "sources", "shape-up-book.md") @@ -142,13 +141,11 @@ func TestStart_MovesToFailedOnError(t *testing.T) { } require.True(t, found, "file should be copied to failed/") - // Original file should still exist (copy, not move — keeps Obsidian vault intact). + // Inbox zero: a quarantined original leaves raw/ too, no marker beside it. _, err := os.Stat(rawFile) - assert.NoError(t, err, "original file should remain in raw/") - - // A .failed marker should exist next to the original. + assert.ErrorIs(t, err, os.ErrNotExist, "original should be removed from raw/ once quarantined") _, err = os.Stat(rawFile + ".failed") - assert.NoError(t, err, ".failed marker should be written") + assert.ErrorIs(t, err, os.ErrNotExist, "no .failed marker: failed/ is the record") // log.md should contain a watcher error entry. logContent, err := os.ReadFile(filepath.Join(brainDir, "log.md")) @@ -390,3 +387,86 @@ func TestProcessDir_DoesNotHangUnderAFIFOSwapRace(t *testing.T) { t.Fatalf("the watcher hung on a swapped-in FIFO after %d iterations", iters.Load()) } } + +// A note whose removal AND marker write both fail (read-only raw/) must not be +// re-ingested on every poll: each pass would write another archive copy. +func TestProcessDir_DoesNotReingestANoteItCannotRetire(t *testing.T) { + brainDir := setupBrainDir(t) + rawDir := filepath.Join(brainDir, "raw") + require.NoError(t, os.WriteFile(filepath.Join(rawDir, "stuck.md"), []byte("Content."), 0o644)) + // processed/ must exist and stay writable; only raw/ itself is locked. + require.NoError(t, os.MkdirAll(filepath.Join(rawDir, "processed"), 0o755)) + require.NoError(t, os.Chmod(rawDir, 0o555)) + t.Cleanup(func() { _ = os.Chmod(rawDir, 0o755) }) + + var calls atomic.Int32 + cfg := Config{ + BrainDir: brainDir, + Pipeline: pipeline.Config{ + Complete: func(ctx context.Context, s, u string) (string, error) { + calls.Add(1) + return successComplete(pipeline.RawPage{Title: "Stuck", Type: "source", Subtype: "article", Content: "x\n"})(ctx, s, u) + }, + Schema: "# Schema", + }, + stuck: &sync.Map{}, + } + for i := 0; i < 3; i++ { + processDir(context.Background(), cfg, "2026-10-19") + } + assert.Equal(t, int32(1), calls.Load(), "a note that cannot be retired is ingested once, not once per poll") +} + +// If the archive copy cannot be written, the original is the only copy left: +// it must stay in raw/. +func TestProcessDir_KeepsTheOriginalWhenTheArchiveCopyFails(t *testing.T) { + brainDir := setupBrainDir(t) + rawDir := filepath.Join(brainDir, "raw") + note := filepath.Join(rawDir, "keep-me.md") + require.NoError(t, os.WriteFile(note, []byte("Content."), 0o644)) + // The dated archive dir exists but is read-only: MkdirAll succeeds, the + // copy itself (os.Create) fails. + dated := filepath.Join(rawDir, "processed", "2026-10-19") + require.NoError(t, os.MkdirAll(dated, 0o755)) + require.NoError(t, os.Chmod(dated, 0o555)) + t.Cleanup(func() { _ = os.Chmod(dated, 0o755) }) + + cfg := Config{ + BrainDir: brainDir, + Pipeline: pipeline.Config{ + Complete: successComplete(pipeline.RawPage{Title: "Keep Me", Type: "source", Subtype: "article", Content: "x\n"}), + Schema: "# Schema", + }, + } + errs := processDir(context.Background(), cfg, "2026-10-19") + + assert.NotEmpty(t, errs, "the failed archive is reported") + _, 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 +// wiki page was extracted from the old content; removing the original would +// leave the new content only in the archive and never re-ingest it. Observes +// size/mtime change only: a same-size rewrite with a restored mtime is not caught. +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)) +}