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

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