diff --git a/ingestion/internal/watcher/watcher.go b/ingestion/internal/watcher/watcher.go index 204da31..5feb3fd 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 @@ -157,7 +172,7 @@ 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) } - retire(root, rel, path+".failed", runErr.Error()) + retire(cfg, root, rel, path, ".failed", runErr.Error()) slog.Warn("watcher: file failed", "file", filename, "error", runErr) @@ -183,7 +198,7 @@ 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) } - retire(root, rel, path+".processed", date) + retire(cfg, root, rel, path, ".processed", date) slog.Info("watcher: file processed", "file", filename, "source", source) return nil @@ -191,14 +206,17 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d // 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(root *os.Root, rel, marker, markerBody string) { +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(marker, []byte(markerBody), 0o644); err != nil { - slog.Error("watcher: failed to write marker", "marker", marker, "error", err) + 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{}{}) + } } } diff --git a/ingestion/internal/watcher/watcher_test.go b/ingestion/internal/watcher/watcher_test.go index 4e2c478..9af0e7d 100644 --- a/ingestion/internal/watcher/watcher_test.go +++ b/ingestion/internal/watcher/watcher_test.go @@ -387,3 +387,60 @@ 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") +}