From 77c44f298c60d69f17d436ad77f74408b3089130 Mon Sep 17 00:00:00 2001 From: Mathias Date: Mon, 5 Oct 2026 22:54:53 +0200 Subject: [PATCH 1/6] fix: pipeline, backfill, watcher and archive writes go through the brain root (#158 part B) The remaining writers created or rewrote brain files by path: - pipeline buildAndWrite: os.MkdirAll + os.WriteFile of every page. - BackfillRefs: os.WriteFile of each concept page; a page swapped for a symlink after the read was written through (the race test overwrote an in-brain target before this change). - watcher: the raw/processed and raw/failed archive copies (os.Create), the .processed/.failed markers and the log.md entry (os.OpenFile). - cmd/server: the claude-session archive. All now go through brainfs PutRegular/AppendRegular (new package-level PutFile/AppendFile for one-off writers: the watcher's call-site lint bans .Open calls in that file so raw/ reads stay on OpenRegular). Stacked on #189 (needs PutRegular/AppendRegular). Co-Authored-By: Claude Opus 5.5 --- ingestion/cmd/server/main.go | 13 ++-- ingestion/internal/brainfs/brainfs.go | 21 ++++++ ingestion/internal/pipeline/backfill.go | 5 +- ingestion/internal/pipeline/pipeline.go | 21 ++++-- ingestion/internal/pipeline/root_test.go | 25 +++++++ ingestion/internal/watcher/watcher.go | 55 +++++++--------- .../internal/watcher/writes_redirect_test.go | 66 +++++++++++++++++++ 7 files changed, 158 insertions(+), 48 deletions(-) create mode 100644 ingestion/internal/watcher/writes_redirect_test.go diff --git a/ingestion/cmd/server/main.go b/ingestion/cmd/server/main.go index db721e4..62dd95a 100644 --- a/ingestion/cmd/server/main.go +++ b/ingestion/cmd/server/main.go @@ -17,6 +17,7 @@ import ( "github.com/mathiasbq/hyperguild/ingestion/internal/api" "github.com/mathiasbq/hyperguild/ingestion/internal/audit" + "github.com/mathiasbq/hyperguild/ingestion/internal/brainfs" "github.com/mathiasbq/hyperguild/ingestion/internal/capture" "github.com/mathiasbq/hyperguild/ingestion/internal/capturehttp" "github.com/mathiasbq/hyperguild/ingestion/internal/classification" @@ -89,11 +90,15 @@ func (s *claudeSink) Ingest(ctx context.Context, b claudewatcher.Batch) error { sb.WriteString("\n\n") } slug := "session-" + b.Host + "-" + b.SessionID - dest := filepath.Join(s.brainDir, "archive", "claude-sessions", b.Host, slug+".md") - if err := os.MkdirAll(filepath.Dir(dest), 0o755); err != nil { - return fmt.Errorf("create claude-sessions archive dir: %w", err) + dest := "archive/claude-sessions/" + b.Host + "/" + slug + ".md" + // Written through the brain root, never through a symlinked directory + // or file (hyperguild#158). + bfs, err := brainfs.Open(s.brainDir) + if err != nil { + return fmt.Errorf("open brain: %w", err) } - if err := os.WriteFile(dest, []byte(sb.String()), 0o644); err != nil { + defer func() { _ = bfs.Close() }() + if err := bfs.PutRegular(dest, []byte(sb.String())); err != nil { return fmt.Errorf("write claude session archive: %w", err) } s.logger.Debug("claude session archived (non-indexed)", "path", dest) diff --git a/ingestion/internal/brainfs/brainfs.go b/ingestion/internal/brainfs/brainfs.go index f7f423d..d8ebeac 100644 --- a/ingestion/internal/brainfs/brainfs.go +++ b/ingestion/internal/brainfs/brainfs.go @@ -243,6 +243,27 @@ func (b *Brain) RealParent(rel string) error { return b.realDir(path.Dir(rel)) } +// PutFile opens the brain dir as a root and calls PutRegular, for one-off +// writers that hold no Brain. +func PutFile(brainDir, rel string, data []byte) error { + b, err := Open(brainDir) + if err != nil { + return err + } + defer func() { _ = b.Close() }() + return b.PutRegular(rel, data) +} + +// AppendFile opens the brain dir as a root and calls AppendRegular. +func AppendFile(brainDir, rel string, data []byte) error { + b, err := Open(brainDir) + if err != nil { + return err + } + defer func() { _ = b.Close() }() + return b.AppendRegular(rel, data) +} + // mkdirReal makes sure every component of rel exists as a real directory, // creating missing ones through the root; an existing component that is // not a real directory is ErrNotRegular. diff --git a/ingestion/internal/pipeline/backfill.go b/ingestion/internal/pipeline/backfill.go index 1217f57..8774f9a 100644 --- a/ingestion/internal/pipeline/backfill.go +++ b/ingestion/internal/pipeline/backfill.go @@ -7,8 +7,6 @@ import ( "fmt" "github.com/mathiasbq/hyperguild/ingestion/internal/brainfs" "io/fs" - "os" - "path/filepath" "strings" "github.com/mathiasbq/hyperguild/ingestion/internal/wiki" @@ -90,8 +88,7 @@ func BackfillRefs(ctx context.Context, brainDir string) (int, error) { } for relPath, page := range pending { - dest := filepath.Join(brainDir, filepath.FromSlash(relPath)) - if err := os.WriteFile(dest, []byte(page.Content), 0o644); err != nil { + if err := bfs.PutRegular(relPath, []byte(page.Content)); err != nil { return 0, fmt.Errorf("write %s: %w", relPath, err) } } diff --git a/ingestion/internal/pipeline/pipeline.go b/ingestion/internal/pipeline/pipeline.go index c368d7f..76f97fb 100644 --- a/ingestion/internal/pipeline/pipeline.go +++ b/ingestion/internal/pipeline/pipeline.go @@ -9,8 +9,6 @@ import ( "fmt" "github.com/mathiasbq/hyperguild/ingestion/internal/brainfs" "io/fs" - "os" - "path/filepath" "strings" "time" @@ -128,10 +126,6 @@ func buildAndWrite(ctx context.Context, rawPages []RawPage, sourceSlug, date, br var written []string for _, page := range merged { if !dryRun { - dest := filepath.Join(brainDir, filepath.FromSlash(page.Path)) - if err := os.MkdirAll(filepath.Dir(dest), 0o755); err != nil { - return Result{}, fmt.Errorf("mkdir for %s: %w", page.Path, err) - } content, conflicts, err := reviseOrCreate(brainDir, page) if err != nil { return Result{}, err @@ -150,7 +144,9 @@ func buildAndWrite(ctx context.Context, rawPages []RawPage, sourceSlug, date, br if err != nil { return Result{}, fmt.Errorf("stamp %s: %w", page.Path, err) } - if err := os.WriteFile(dest, []byte(stamped), 0o644); err != nil { + // Written through the brain root, never through a symlinked + // directory or page (hyperguild#158). + if err := writeBrainFile(brainDir, page.Path, []byte(stamped)); err != nil { return Result{}, fmt.Errorf("write %s: %w", page.Path, err) } } @@ -201,6 +197,17 @@ func readSourcePage(brainDir, sourceSlug string) string { return string(raw) } +// writeBrainFile creates or replaces one brain-relative file through the +// brain root (hyperguild#158). +func writeBrainFile(brainDir, rel string, data []byte) error { + bfs, err := brainfs.Open(brainDir) + if err != nil { + return err + } + defer func() { _ = bfs.Close() }() + return bfs.PutRegular(rel, data) +} + // readBrainFile reads one brain-relative file through the brain root, // regular files only (hyperguild#157). func readBrainFile(brainDir, rel string) ([]byte, error) { diff --git a/ingestion/internal/pipeline/root_test.go b/ingestion/internal/pipeline/root_test.go index b9f7b6c..d6f4fba 100644 --- a/ingestion/internal/pipeline/root_test.go +++ b/ingestion/internal/pipeline/root_test.go @@ -165,3 +165,28 @@ func TestBackfillRefsDoesNotCopyAPageSwappedInBehindASymlink(t *testing.T) { require.Contains(t, read(), "[[s|S]]", "positive control: the regular page gets the source ref") brainfstest.AssertNoLeakFrom(t, page, inside, target, 3*time.Second, read) } + +// hyperguild#158: BackfillRefs read a regular concept page, a symlink could +// be swapped in, and os.WriteFile then followed it and overwrote the link's +// target. The target is not restored between rounds here, so any +// write-through shows. +func TestBackfillRefsNeverWritesThroughAPageSwappedForASymlink(t *testing.T) { + brain := t.TempDir() + mkdirs(t, brain, "wiki/sources", "wiki/concepts", "raw") + require.NoError(t, os.WriteFile(filepath.Join(brain, "wiki", "sources", "s.md"), []byte("---\ntitle: S\n---\n[[c|C]]\n"), 0o644)) + page := filepath.Join(brain, "wiki", "concepts", "c.md") + target := filepath.Join(brain, "raw", "secret.md") + require.NoError(t, os.WriteFile(target, []byte("untouched"), 0o644)) + inside := "---\ntitle: C\n---\nbody\n" + require.NoError(t, os.WriteFile(page, []byte(inside), 0o644)) + brainfstest.StartSwap(t, page, inside, target, false) + ok := 0 + for deadline := time.Now().Add(2 * time.Second); time.Now().Before(deadline); { + if _, err := BackfillRefs(context.Background(), brain); err == nil { + ok++ + } + } + got, _ := os.ReadFile(target) + assert.Equal(t, "untouched", string(got), "BackfillRefs wrote through a swapped-in symlink") + assert.Positive(t, ok, "positive control: some backfills ran") +} diff --git a/ingestion/internal/watcher/watcher.go b/ingestion/internal/watcher/watcher.go index b5c41b3..069590f 100644 --- a/ingestion/internal/watcher/watcher.go +++ b/ingestion/internal/watcher/watcher.go @@ -12,6 +12,7 @@ import ( "time" "unicode" + "github.com/mathiasbq/hyperguild/ingestion/internal/brainfs" "github.com/mathiasbq/hyperguild/ingestion/internal/extract" "github.com/mathiasbq/hyperguild/ingestion/internal/pipeline" ) @@ -145,16 +146,12 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d _, runErr := pipeline.Run(ctx, cfg.Pipeline, cfg.BrainDir, content, source, false) if runErr != nil { - // Copy to failed/ and leave a .failed marker so we don't retry. - failedDir := filepath.Join(cfg.BrainDir, "raw", "failed") - if mkErr := os.MkdirAll(failedDir, 0o755); mkErr != nil { - return fmt.Errorf("mkdir failed dir: %w", mkErr) - } - dest := filepath.Join(failedDir, filename) - if cpErr := copyFromRoot(root, rel, dest); cpErr != nil { + // Copy to failed/ and leave a .failed marker so we don't retry. Both + // written through the brain root (hyperguild#158). + if cpErr := copyFromRoot(root, rel, cfg.BrainDir, "raw/failed/"+filename); cpErr != nil { return fmt.Errorf("copy to failed: %w", cpErr) } - if mkErr := os.WriteFile(path+".failed", []byte(runErr.Error()), 0o644); mkErr != nil { + if mkErr := putBrainFile(cfg.BrainDir, "raw/"+filepath.ToSlash(rel)+".failed", []byte(runErr.Error())); mkErr != nil { slog.Error("watcher: failed to write .failed marker", "error", mkErr) } @@ -168,21 +165,18 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d } // Copy to processed/YYYY-MM-DD/ and leave a .processed marker so we don't re-ingest. - processedDir := filepath.Join(cfg.BrainDir, "raw", "processed", date) - if err := os.MkdirAll(processedDir, 0o755); err != nil { - return fmt.Errorf("mkdir processed dir: %w", err) - } - dest := filepath.Join(processedDir, filename) - if _, err := os.Stat(dest); err == nil { + // Written through the brain root (hyperguild#158). + dest := "raw/processed/" + date + "/" + filename + if _, err := os.Lstat(filepath.Join(cfg.BrainDir, filepath.FromSlash(dest))); err == nil { // Archive copy already exists; append timestamp to avoid overwriting. ext := filepath.Ext(filename) base := strings.TrimSuffix(filename, ext) - dest = filepath.Join(processedDir, base+"-"+time.Now().UTC().Format("150405")+ext) + dest = "raw/processed/" + date + "/" + base + "-" + time.Now().UTC().Format("150405") + ext } - if err := copyFromRoot(root, rel, dest); err != nil { + if err := copyFromRoot(root, rel, cfg.BrainDir, dest); err != nil { return fmt.Errorf("copy to processed: %w", err) } - if err := os.WriteFile(path+".processed", []byte(date), 0o644); err != nil { + if err := putBrainFile(cfg.BrainDir, "raw/"+filepath.ToSlash(rel)+".processed", []byte(date)); err != nil { slog.Error("watcher: failed to write .processed marker", "error", err) } @@ -214,23 +208,22 @@ func appendWatcherLog(brainDir, filename string, runErr error, date string) erro entry := fmt.Sprintf("## %s — watcher error\n\n- **File:** %s\n- **Error:** %s\n\n", date, filename, runErr.Error()) - logPath := filepath.Join(brainDir, "log.md") - f, err := os.OpenFile(logPath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644) - if err != nil { - return fmt.Errorf("open log: %w", err) - } - - if _, err = f.WriteString(entry); err != nil { - f.Close() //nolint:errcheck + if err := brainfs.AppendFile(brainDir, "log.md", []byte(entry)); err != nil { return fmt.Errorf("write log: %w", err) } - return f.Close() + return nil +} + +// putBrainFile creates or replaces a brain-relative file through the brain +// root, never through a symlinked directory or file (hyperguild#158). +func putBrainFile(brainDir, rel string, data []byte) error { + return brainfs.PutFile(brainDir, rel, data) } // copyFromRoot archives a raw/ file read through the root, refusing anything // that is not a regular file, so a swapped-in symlink cannot copy an outside // file into raw/processed/ or raw/failed/ (hyperguild#153). -func copyFromRoot(root *os.Root, rel, dst string) error { +func copyFromRoot(root *os.Root, rel, brainDir, dstRel string) error { fi, err := root.Lstat(rel) if err != nil { return err @@ -243,13 +236,9 @@ func copyFromRoot(root *os.Root, rel, dst string) error { return err } defer func() { _ = in.Close() }() - out, err := os.Create(dst) + data, err := io.ReadAll(in) if err != nil { return err } - if _, err := io.Copy(out, in); err != nil { - _ = out.Close() - return err - } - return out.Close() + return putBrainFile(brainDir, dstRel, data) } diff --git a/ingestion/internal/watcher/writes_redirect_test.go b/ingestion/internal/watcher/writes_redirect_test.go new file mode 100644 index 0000000..ed46375 --- /dev/null +++ b/ingestion/internal/watcher/writes_redirect_test.go @@ -0,0 +1,66 @@ +package watcher + +import ( + "context" + "encoding/json" + "errors" + "os" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/mathiasbq/hyperguild/ingestion/internal/pipeline" +) + +func okPipeline() pipeline.Config { + return pipeline.Config{Complete: func(context.Context, string, string) (string, error) { + b, _ := json.Marshal([]pipeline.RawPage{{Title: "X", Type: "source", Subtype: "article", Content: "## S\n\nx\n"}}) + return string(b), nil + }, Schema: "# Schema"} +} + +func failingPipeline() pipeline.Config { + return pipeline.Config{Complete: func(context.Context, string, string) (string, error) { + return "", errors.New("model down") + }, Schema: "# Schema"} +} + +// hyperguild#158: the watcher wrote the archive copy under raw/processed or +// raw/failed, and its log.md entry, by path, so a symlinked archive +// directory or log redirected the write elsewhere in the brain or outside +// it. (A pre-planted marker symlink makes the watcher skip the note, so the +// marker write itself is reachable only by a race.) +func TestProcessDirDoesNotWriteThroughASymlinkedArchiveDirOrLog(t *testing.T) { + for name, pcfg := range map[string]pipeline.Config{"processed": okPipeline(), "failed": failingPipeline()} { + t.Run(name, func(t *testing.T) { + brainDir := setupBrainDir(t) + outside := t.TempDir() + victim := filepath.Join(brainDir, "wiki", "concepts", "v.md") + require.NoError(t, os.WriteFile(victim, []byte("untouched"), 0o644)) + require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "n.md"), []byte("note"), 0o644)) + require.NoError(t, os.Symlink(outside, filepath.Join(brainDir, "raw", name))) + if name == "failed" { + require.NoError(t, os.Symlink(filepath.Join("wiki", "concepts", "v.md"), filepath.Join(brainDir, "log.md"))) + } + _ = processDir(context.Background(), Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: pcfg}, "2026-10-05") + got, _ := os.ReadFile(victim) + assert.Equal(t, "untouched", string(got), "the log was written through a symlink") + ents, _ := os.ReadDir(outside) + assert.Empty(t, ents, "the archive copy was written outside the brain") + }) + } +} + +func TestProcessDirWritesMarkersAndArchiveInRealDirectories(t *testing.T) { + brainDir := setupBrainDir(t) + require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "n.md"), []byte("note"), 0o644)) + errs := processDir(context.Background(), Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: okPipeline()}, "2026-10-05") + assert.Empty(t, errs) + _, err := os.Stat(filepath.Join(brainDir, "raw", "n.md.processed")) + assert.NoError(t, err, "positive control: marker written") + _, err = os.Stat(filepath.Join(brainDir, "raw", "processed", "2026-10-05", "n.md")) + assert.NoError(t, err, "positive control: archive copy written") +} From 1a2a93d2f486e941ad0f00fc61e0d384d3293f1f Mon Sep 17 00:00:00 2001 From: Mathias Date: Mon, 5 Oct 2026 23:40:19 +0200 Subject: [PATCH 2/6] fix(watcher): refusals end in a terminal state; pin every converted writer (#158) Review round 1 on PR #195: - A refused marker or archive write left no marker, so the note was reprocessed every interval (archive copies piling up; an LLM call per round on the failure path). The marker check now uses Lstat (any entry at the marker path counts, so a dangling marker link skips the note instead of looping), and a note whose archive copy or marker write is refused is remembered (path, size, mtime) and not retried until it changes. - cleanRel no longer refuses a backslash: it is a legal filename byte on Linux, not a traversal, and a raw note named with one was never archived or marked. - Tests now reach every converted writer: the log.md append (real raw/failed so the log is reached, symlinked log.md), a dangling marker link, the refused-retry loop with a positive control, a backslash name, a buildAndWrite swap race, and the claude-session archive with in-brain and outside symlinked archive dirs. The stale lint allowlist entry for appendWatcherLog is removed. Co-Authored-By: Claude Opus 5.5 --- ingestion/cmd/server/main_test.go | 24 ++++++ ingestion/internal/brainfs/brainfs.go | 2 +- ingestion/internal/extract/callsites_test.go | 1 - ingestion/internal/pipeline/root_test.go | 25 +++++++ ingestion/internal/watcher/watcher.go | 52 ++++++++++++- .../internal/watcher/writes_redirect_test.go | 74 +++++++++++++++++-- 6 files changed, 169 insertions(+), 9 deletions(-) diff --git a/ingestion/cmd/server/main_test.go b/ingestion/cmd/server/main_test.go index 4d1236d..b52f0c7 100644 --- a/ingestion/cmd/server/main_test.go +++ b/ingestion/cmd/server/main_test.go @@ -76,3 +76,27 @@ func TestAlertingWarnings(t *testing.T) { }) } } + +// hyperguild#158: the claude-session archive was written with os.MkdirAll + +// os.WriteFile by path, so a symlinked archive directory redirected it. +func TestClaudeSinkDoesNotWriteThroughASymlinkedArchiveDir(t *testing.T) { + for name, mk := range map[string]func(t *testing.T, dir string) (link, target string){ + "outside": func(t *testing.T, _ string) (string, string) { o := t.TempDir(); return o, o }, + "in-brain": func(t *testing.T, dir string) (string, string) { + require.NoError(t, os.MkdirAll(filepath.Join(dir, "raw"), 0o755)) + return filepath.Join("..", "raw"), filepath.Join(dir, "raw") + }, + } { + t.Run(name, func(t *testing.T) { + dir := t.TempDir() + link, target := mk(t, dir) + require.NoError(t, os.Symlink(link, filepath.Join(dir, "archive"))) + sink := &claudeSink{brainDir: dir, logger: slog.New(slog.NewTextHandler(os.Stderr, nil))} + err := sink.Ingest(context.Background(), claudewatcher.Batch{Host: "koala", SessionID: "abc", + Turns: []claudewatcher.Turn{{Type: "assistant", Content: "x"}}}) + assert.Error(t, err) + ents, _ := os.ReadDir(target) + assert.Empty(t, ents, "the archive was written at the link's target") + }) + } +} diff --git a/ingestion/internal/brainfs/brainfs.go b/ingestion/internal/brainfs/brainfs.go index d8ebeac..f083f50 100644 --- a/ingestion/internal/brainfs/brainfs.go +++ b/ingestion/internal/brainfs/brainfs.go @@ -109,7 +109,7 @@ func (b *Brain) ReadDirRegular(relDir string) ([]string, error) { // after following a symlink, and a ".." after a symlinked directory would // otherwise pass the check and redirect the read or write. func cleanRel(rel string) error { - if rel == "" || path.Clean(rel) != rel || !filepath.IsLocal(rel) || strings.Contains(rel, "\\") { + if rel == "" || path.Clean(rel) != rel || !filepath.IsLocal(rel) { return fmt.Errorf("%q: not a clean brain-relative path: %w", rel, ErrNotRegular) } return nil diff --git a/ingestion/internal/extract/callsites_test.go b/ingestion/internal/extract/callsites_test.go index c544e2e..293ec54 100644 --- a/ingestion/internal/extract/callsites_test.go +++ b/ingestion/internal/extract/callsites_test.go @@ -88,7 +88,6 @@ func TestNoPackageReferencesExtractTextOutsideExtract(t *testing.T) { // osOpenAllowed names the functions in guarded files that may call os.Open* // on a path that is not a raw/ read. var osOpenAllowed = map[string]bool{ - "internal/watcher/watcher.go:appendWatcherLog": true, // ingest log append, not a raw/ read // Text reads by path; outside this package only TextIn's temp copy reaches // it (TestNoPackageReferencesExtractTextOutsideExtract enforces that). "internal/extract/extract.go:Text": true, diff --git a/ingestion/internal/pipeline/root_test.go b/ingestion/internal/pipeline/root_test.go index d6f4fba..a727e2d 100644 --- a/ingestion/internal/pipeline/root_test.go +++ b/ingestion/internal/pipeline/root_test.go @@ -190,3 +190,28 @@ func TestBackfillRefsNeverWritesThroughAPageSwappedForASymlink(t *testing.T) { assert.Equal(t, "untouched", string(got), "BackfillRefs wrote through a swapped-in symlink") assert.Positive(t, ok, "positive control: some backfills ran") } + +// hyperguild#158: buildAndWrite wrote each page with os.WriteFile by path, so +// a page read as regular and swapped for a symlink before the write was +// written through. The target is not restored between rounds, so any +// write-through shows. +func TestBuildAndWriteNeverWritesThroughAPageSwappedForASymlink(t *testing.T) { + brain := t.TempDir() + mkdirs(t, brain, "wiki/sources", "wiki/concepts", "wiki/entities", "raw") + page := filepath.Join(brain, "wiki", "concepts", "zq-concept.md") + target := filepath.Join(brain, "raw", "secret.md") + require.NoError(t, os.WriteFile(target, []byte("untouched"), 0o644)) + inside := "---\ntitle: Zq Concept\n---\nbody\n" + require.NoError(t, os.WriteFile(page, []byte(inside), 0o644)) + pages := []RawPage{{Title: "Zq Concept", Type: "concept", Content: "## About\n\nmore\n"}} + brainfstest.StartSwap(t, page, inside, target, false) + ok := 0 + for deadline := time.Now().Add(2 * time.Second); time.Now().Before(deadline); { + if _, err := RunRaw(context.Background(), brain, "zq source", pages, false); err == nil { + ok++ + } + } + got, _ := os.ReadFile(target) + assert.Equal(t, "untouched", string(got), "buildAndWrite wrote through a swapped-in symlink") + assert.Positive(t, ok, "positive control: some runs wrote the page") +} diff --git a/ingestion/internal/watcher/watcher.go b/ingestion/internal/watcher/watcher.go index 069590f..348eb47 100644 --- a/ingestion/internal/watcher/watcher.go +++ b/ingestion/internal/watcher/watcher.go @@ -3,12 +3,14 @@ package watcher import ( "context" + "errors" "fmt" "io" "log/slog" "os" "path/filepath" "strings" + "sync" "time" "unicode" @@ -110,14 +112,28 @@ func processDir(ctx context.Context, cfg Config, date string) []error { } // Skip files that have already been processed or permanently failed. - if _, err := os.Stat(path + ".processed"); err == nil { + // Any entry at the marker path counts (Lstat, not Stat): a dangling + // marker symlink made os.Stat report nothing and the note was + // reprocessed every interval (hyperguild#158). + if _, err := os.Lstat(path + ".processed"); err == nil { return nil } - if _, err := os.Stat(path + ".failed"); err == nil { + if _, err := os.Lstat(path + ".failed"); err == nil { + return nil + } + // A note whose archive copy or marker was refused is not retried + // until it changes: a refusal has no marker to stop the next round, + // and each round can cost an LLM call (hyperguild#158). + info, ierr := d.Info() + if ierr == nil && refusedUnchanged(path, info) { return nil } if err := processFile(ctx, cfg, root, rawDir, path, date); err != nil { + if ierr == nil && errors.Is(err, brainfs.ErrNotRegular) { + rememberRefused(path, info) + slog.Error("watcher: note refused; not retried until it changes", "file", filepath.Base(path), "error", err) + } errs = append(errs, fmt.Errorf("process %s: %w", filepath.Base(path), err)) } return nil @@ -152,6 +168,9 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d return fmt.Errorf("copy to failed: %w", cpErr) } if mkErr := putBrainFile(cfg.BrainDir, "raw/"+filepath.ToSlash(rel)+".failed", []byte(runErr.Error())); mkErr != nil { + if errors.Is(mkErr, brainfs.ErrNotRegular) { + return fmt.Errorf("write .failed marker: %w", mkErr) + } slog.Error("watcher: failed to write .failed marker", "error", mkErr) } @@ -177,6 +196,9 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d return fmt.Errorf("copy to processed: %w", err) } if err := putBrainFile(cfg.BrainDir, "raw/"+filepath.ToSlash(rel)+".processed", []byte(date)); err != nil { + if errors.Is(err, brainfs.ErrNotRegular) { + return fmt.Errorf("write .processed marker: %w", err) + } slog.Error("watcher: failed to write .processed marker", "error", err) } @@ -184,6 +206,32 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d return nil } +// refused remembers notes whose archive copy or marker write was refused, +// by path, with the size and mtime they had, for the life of the process. +var refused sync.Map + +type refusedStamp struct { + size int64 + mtime time.Time +} + +func rememberRefused(path string, info os.FileInfo) { + refused.Store(path, refusedStamp{size: info.Size(), mtime: info.ModTime()}) +} + +func refusedUnchanged(path string, info os.FileInfo) bool { + v, ok := refused.Load(path) + if !ok { + return false + } + st := v.(refusedStamp) + if st.size == info.Size() && st.mtime.Equal(info.ModTime()) { + return true + } + refused.Delete(path) + return false +} + // 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/writes_redirect_test.go b/ingestion/internal/watcher/writes_redirect_test.go index ed46375..fe4efed 100644 --- a/ingestion/internal/watcher/writes_redirect_test.go +++ b/ingestion/internal/watcher/writes_redirect_test.go @@ -33,7 +33,7 @@ func failingPipeline() pipeline.Config { // directory or log redirected the write elsewhere in the brain or outside // it. (A pre-planted marker symlink makes the watcher skip the note, so the // marker write itself is reachable only by a race.) -func TestProcessDirDoesNotWriteThroughASymlinkedArchiveDirOrLog(t *testing.T) { +func TestProcessDirDoesNotWriteThroughASymlinkedArchiveDir(t *testing.T) { for name, pcfg := range map[string]pipeline.Config{"processed": okPipeline(), "failed": failingPipeline()} { t.Run(name, func(t *testing.T) { brainDir := setupBrainDir(t) @@ -42,12 +42,9 @@ func TestProcessDirDoesNotWriteThroughASymlinkedArchiveDirOrLog(t *testing.T) { require.NoError(t, os.WriteFile(victim, []byte("untouched"), 0o644)) require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "n.md"), []byte("note"), 0o644)) require.NoError(t, os.Symlink(outside, filepath.Join(brainDir, "raw", name))) - if name == "failed" { - require.NoError(t, os.Symlink(filepath.Join("wiki", "concepts", "v.md"), filepath.Join(brainDir, "log.md"))) - } _ = processDir(context.Background(), Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: pcfg}, "2026-10-05") got, _ := os.ReadFile(victim) - assert.Equal(t, "untouched", string(got), "the log was written through a symlink") + assert.Equal(t, "untouched", string(got)) ents, _ := os.ReadDir(outside) assert.Empty(t, ents, "the archive copy was written outside the brain") }) @@ -64,3 +61,70 @@ func TestProcessDirWritesMarkersAndArchiveInRealDirectories(t *testing.T) { _, err = os.Stat(filepath.Join(brainDir, "raw", "processed", "2026-10-05", "n.md")) assert.NoError(t, err, "positive control: archive copy written") } + +// The log.md entry for a failed note is appended through the brain root: with +// a real raw/failed (so the archive copy succeeds and the log is reached), a +// symlinked log.md is not written through. +func TestProcessDirDoesNotAppendTheLogThroughASymlink(t *testing.T) { + brainDir := setupBrainDir(t) + victim := filepath.Join(brainDir, "wiki", "concepts", "v.md") + require.NoError(t, os.WriteFile(victim, []byte("untouched"), 0o644)) + require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "n.md"), []byte("note"), 0o644)) + require.NoError(t, os.Symlink(filepath.Join("wiki", "concepts", "v.md"), filepath.Join(brainDir, "log.md"))) + _ = processDir(context.Background(), Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: failingPipeline()}, "2026-10-05") + _, err := os.Stat(filepath.Join(brainDir, "raw", "failed", "n.md")) + require.NoError(t, err, "positive control: the archive copy was written, so the log append was reached") + got, _ := os.ReadFile(victim) + assert.Equal(t, "untouched", string(got), "the log was appended through a symlink") +} + +// A marker path holding anything (here a dangling symlink) counts as a +// marker: the note is skipped and nothing is created at the link's target. +// (os.Stat followed the link, saw nothing, and processed the note again.) +func TestProcessDirTreatsAnyEntryAtTheMarkerPathAsAMarker(t *testing.T) { + brainDir := setupBrainDir(t) + outside := filepath.Join(t.TempDir(), "created") + require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "n.md"), []byte("note"), 0o644)) + require.NoError(t, os.Symlink(outside, filepath.Join(brainDir, "raw", "n.md.processed"))) + calls := 0 + cfg := Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: pipeline.Config{ + Complete: func(context.Context, string, string) (string, error) { calls++; return "[]", nil }, Schema: "# Schema"}} + _ = processDir(context.Background(), cfg, "2026-10-05") + assert.Zero(t, calls, "the note was processed despite an entry at its marker path") + _, err := os.Lstat(outside) + assert.True(t, os.IsNotExist(err), "a file was created at the marker link's target") +} + +// A refused archive copy must not retry every interval: the note is +// remembered and skipped until it changes (a symlinked raw/failed cost one +// LLM call per round before). +func TestProcessDirDoesNotRetryANoteWhoseArchiveCopyIsRefused(t *testing.T) { + brainDir := setupBrainDir(t) + require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "n.md"), []byte("note"), 0o644)) + require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(brainDir, "raw", "failed"))) + calls := 0 + cfg := Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: pipeline.Config{ + Complete: func(context.Context, string, string) (string, error) { calls++; return "", errors.New("model down") }, Schema: "# Schema"}} + for i := 0; i < 3; i++ { + _ = processDir(context.Background(), cfg, "2026-10-05") + } + assert.Equal(t, 1, calls, "the refused note was retried every round") + + // Once the note changes it is tried again. + require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "n.md"), []byte("note, edited"), 0o644)) + future := time.Now().Add(2 * time.Second) + require.NoError(t, os.Chtimes(filepath.Join(brainDir, "raw", "n.md"), future, future)) + _ = processDir(context.Background(), cfg, "2026-10-05") + assert.Equal(t, 2, calls, "positive control: a changed note is tried again") +} + +// A note whose name has a backslash is legal on Linux and must be processed +// (cleanRel refused it, so it was never archived or marked). +func TestProcessDirProcessesANoteWithABackslashInItsName(t *testing.T) { + brainDir := setupBrainDir(t) + require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "a\\b.md"), []byte("note"), 0o644)) + errs := processDir(context.Background(), Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: okPipeline()}, "2026-10-05") + assert.Empty(t, errs) + _, err := os.Stat(filepath.Join(brainDir, "raw", "a\\b.md.processed")) + assert.NoError(t, err) +} From 22d0653b6196146f3f8c1c9e57afeadf83f4b52e Mon Sep 17 00:00:00 2001 From: Mathias Date: Mon, 5 Oct 2026 23:47:24 +0200 Subject: [PATCH 3/6] test(watcher): a marker link planted during the run is refused and not retried (#158) Review round 2 on PR #195: the marker writes had no test, so reverting them to by-path writes, or the refusal to log-only, stayed green. The link is planted from inside the fake model call, between the walk's Lstat and the marker write, for both markers and both an outside and an in-brain relative target; red on all four flips. The marker-entry skip is now checked for .failed too. Also pins cleanRel's acceptance of a backslash as a filename byte in brainfs itself, and stops the buildAndWrite control claiming writes it does not count. Co-Authored-By: Claude Opus 5.5 --- ingestion/internal/brainfs/brainfs_test.go | 17 ++++ ingestion/internal/pipeline/root_test.go | 2 +- .../internal/watcher/writes_redirect_test.go | 78 ++++++++++++++++--- 3 files changed, 85 insertions(+), 12 deletions(-) diff --git a/ingestion/internal/brainfs/brainfs_test.go b/ingestion/internal/brainfs/brainfs_test.go index 0f5570b..5ac4f4d 100644 --- a/ingestion/internal/brainfs/brainfs_test.go +++ b/ingestion/internal/brainfs/brainfs_test.go @@ -529,3 +529,20 @@ func TestCreateRegularCreatesOnlyNewFilesThroughRealDirectories(t *testing.T) { got, _ = os.ReadFile(filepath.Join(dir, "wiki", "new", "facts", "n.md")) assert.Equal(t, "one", string(got)) } + +// On Linux a backslash is an ordinary filename byte, not a separator, so a +// name containing one stays a single in-root component (hyperguild#158: the +// watcher retried such a note forever while cleanRel refused it). +func TestReadAndPutRegularTreatABackslashAsAFilenameByte(t *testing.T) { + dir := t.TempDir() + b, err := Open(dir) + require.NoError(t, err) + defer func() { _ = b.Close() }() + require.NoError(t, b.PutRegular(`a\..\b.md`, []byte("x"))) + got, err := os.ReadFile(filepath.Join(dir, `a\..\b.md`)) + require.NoError(t, err, "the name was not kept as one component") + assert.Equal(t, "x", string(got)) + got, err = b.ReadRegular(`a\..\b.md`) + require.NoError(t, err) + assert.Equal(t, "x", string(got)) +} diff --git a/ingestion/internal/pipeline/root_test.go b/ingestion/internal/pipeline/root_test.go index a727e2d..9b92cde 100644 --- a/ingestion/internal/pipeline/root_test.go +++ b/ingestion/internal/pipeline/root_test.go @@ -213,5 +213,5 @@ func TestBuildAndWriteNeverWritesThroughAPageSwappedForASymlink(t *testing.T) { } got, _ := os.ReadFile(target) assert.Equal(t, "untouched", string(got), "buildAndWrite wrote through a swapped-in symlink") - assert.Positive(t, ok, "positive control: some runs wrote the page") + assert.Positive(t, ok, "positive control: some runs returned without error") } diff --git a/ingestion/internal/watcher/writes_redirect_test.go b/ingestion/internal/watcher/writes_redirect_test.go index fe4efed..dc06188 100644 --- a/ingestion/internal/watcher/writes_redirect_test.go +++ b/ingestion/internal/watcher/writes_redirect_test.go @@ -82,17 +82,73 @@ func TestProcessDirDoesNotAppendTheLogThroughASymlink(t *testing.T) { // marker: the note is skipped and nothing is created at the link's target. // (os.Stat followed the link, saw nothing, and processed the note again.) func TestProcessDirTreatsAnyEntryAtTheMarkerPathAsAMarker(t *testing.T) { - brainDir := setupBrainDir(t) - outside := filepath.Join(t.TempDir(), "created") - require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "n.md"), []byte("note"), 0o644)) - require.NoError(t, os.Symlink(outside, filepath.Join(brainDir, "raw", "n.md.processed"))) - calls := 0 - cfg := Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: pipeline.Config{ - Complete: func(context.Context, string, string) (string, error) { calls++; return "[]", nil }, Schema: "# Schema"}} - _ = processDir(context.Background(), cfg, "2026-10-05") - assert.Zero(t, calls, "the note was processed despite an entry at its marker path") - _, err := os.Lstat(outside) - assert.True(t, os.IsNotExist(err), "a file was created at the marker link's target") + for _, kind := range []string{"processed", "failed"} { + t.Run(kind, func(t *testing.T) { + brainDir := setupBrainDir(t) + outside := filepath.Join(t.TempDir(), "created") + require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "n.md"), []byte("note"), 0o644)) + require.NoError(t, os.Symlink(outside, filepath.Join(brainDir, "raw", "n.md."+kind))) + calls := 0 + cfg := Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: pipeline.Config{ + Complete: func(context.Context, string, string) (string, error) { calls++; return "[]", nil }, Schema: "# Schema"}} + _ = processDir(context.Background(), cfg, "2026-10-05") + assert.Zero(t, calls, "the note was processed despite an entry at its marker path") + _, err := os.Lstat(outside) + assert.True(t, os.IsNotExist(err), "a file was created at the marker link's target") + }) + } +} + +// The marker write itself is reachable while pipeline.Run is in flight +// (in production, an LLM call): a link planted at the marker path then is +// refused, and the note is remembered, so removing the link (as +// symlinkguard does) does not make the watcher call the model again. +func TestProcessDirRefusesAMarkerLinkPlantedDuringTheRunAndDoesNotRetry(t *testing.T) { + for _, kind := range []string{"processed", "failed"} { + for _, target := range []string{"outside", "in-brain"} { + t.Run(kind+"/"+target, func(t *testing.T) { + brainDir := setupBrainDir(t) + raw := filepath.Join(brainDir, "raw") + marker := filepath.Join(raw, "n.md."+kind) + require.NoError(t, os.WriteFile(filepath.Join(raw, "n.md"), []byte("note"), 0o644)) + // The in-brain target is relative, so os.Root itself would + // follow it; only the regular-file check refuses it. + victim := filepath.Join(raw, "victim") + link := filepath.Join(t.TempDir(), "created") + if target == "in-brain" { + require.NoError(t, os.WriteFile(victim, []byte("untouched"), 0o644)) + link = "victim" + } + ok := okPipeline().Complete + calls := 0 + cfg := Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: pipeline.Config{Schema: "# Schema", + Complete: func(ctx context.Context, sys, user string) (string, error) { + calls++ + require.NoError(t, os.Symlink(link, marker)) + if kind == "failed" { + return "", errors.New("model down") + } + return ok(ctx, sys, user) + }}} + for i := 0; i < 3; i++ { + errs := processDir(context.Background(), cfg, "2026-10-05") + if i == 0 { + require.NotEmpty(t, errs, "the planted link was not refused") + } + _ = os.Remove(marker) + } + assert.Equal(t, 1, calls, "the note was retried after its marker write was refused") + if target == "in-brain" { + got, err := os.ReadFile(victim) + require.NoError(t, err) + assert.Equal(t, "untouched", string(got), "the marker was written through the link") + } else { + _, err := os.Lstat(link) + assert.True(t, os.IsNotExist(err), "the marker was written through the link") + } + }) + } + } } // A refused archive copy must not retry every interval: the note is From d8c7defc78cb634b393406a6e5d9baa55c943097 Mon Sep 17 00:00:00 2001 From: Mathias Date: Tue, 20 Oct 2026 20:04:02 +0200 Subject: [PATCH 4/6] fix(watcher): refuse oversize notes before reading them; pin the long-name marker case (#198, #199) MaxNoteBytes (default 64 MiB) is checked on the Lstat size before extraction; an oversize note stays in raw/, is reported once and not retried until it changes. The #199 case (marker name over NAME_MAX) is covered by the remember-on-any-failure retire path and now has a test that fails without it. Co-Authored-By: Claude Sonnet 5.5 --- ingestion/internal/watcher/watcher.go | 21 +++++++- ingestion/internal/watcher/watcher_test.go | 57 ++++++++++++++++++++++ 2 files changed, 77 insertions(+), 1 deletion(-) diff --git a/ingestion/internal/watcher/watcher.go b/ingestion/internal/watcher/watcher.go index d7cadb3..14c6280 100644 --- a/ingestion/internal/watcher/watcher.go +++ b/ingestion/internal/watcher/watcher.go @@ -27,12 +27,24 @@ type Config struct { Interval time.Duration Pipeline pipeline.Config + // MaxNoteBytes refuses a note larger than this before anything is read into + // memory (hyperguild#198). 0 means defaultMaxNoteBytes. + MaxNoteBytes int64 + // Resolver, when set, decides each note's wing (hyperguild#205): explicit // subfolder/frontmatter, else classifier + scorer. A note it cannot place // is parked in raw/needs-review/ for a human. Nil keeps the retired layout. Resolver *wing.Resolver } +// defaultMaxNoteBytes bounds a single note: extraction and the archive copy both +// hold it in memory. +const defaultMaxNoteBytes = 64 << 20 + +// errTooLarge marks a note refused for its size; like a refused write it is +// reported once and not retried until the note changes. +var errTooLarge = errors.New("note too large") + // 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) { @@ -137,7 +149,7 @@ func processDir(ctx context.Context, cfg Config, date string) []error { } if err := processFile(ctx, cfg, root, rawDir, path, date); err != nil { - if ierr == nil && errors.Is(err, brainfs.ErrNotRegular) { + if ierr == nil && (errors.Is(err, brainfs.ErrNotRegular) || errors.Is(err, errTooLarge)) { rememberRefused(path, info) slog.Error("watcher: note refused; not retried until it changes", "file", filepath.Base(path), "error", err) } @@ -169,6 +181,13 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d if err != nil { return fmt.Errorf("stat before ingest: %w", err) } + limit := cfg.MaxNoteBytes + if limit <= 0 { + limit = defaultMaxNoteBytes + } + if before.Size() > limit { + return fmt.Errorf("%s is %d bytes, over the %d byte cap: %w", filename, before.Size(), limit, errTooLarge) + } content, err := extract.TextIn(root, rel) if err != nil { return fmt.Errorf("extract text: %w", err) diff --git a/ingestion/internal/watcher/watcher_test.go b/ingestion/internal/watcher/watcher_test.go index f76d562..179d70c 100644 --- a/ingestion/internal/watcher/watcher_test.go +++ b/ingestion/internal/watcher/watcher_test.go @@ -475,3 +475,60 @@ func TestProcessDir_KeepsANoteThatChangedWhileItWasBeingIngested(t *testing.T) { require.NoError(t, err, "a note that changed during ingest must stay in raw/") assert.Equal(t, "newer and longer content", string(got)) } + +// hyperguild#198: an oversized note must not be read into memory. It is left in +// raw/ (visible, not lost), reported once, and not retried until it changes. +func TestProcessDir_ANoteOverTheSizeCapIsRefusedOnceAndKept(t *testing.T) { + brainDir := setupBrainDir(t) + big := filepath.Join(brainDir, "raw", "big.md") + require.NoError(t, os.WriteFile(big, []byte(strings.Repeat("x", 2048)), 0o644)) + var calls atomic.Int32 + cfg := Config{ + BrainDir: brainDir, + MaxNoteBytes: 1024, + Pipeline: pipeline.Config{ + Complete: func(ctx context.Context, s, u string) (string, error) { + calls.Add(1) + return successComplete(pipeline.RawPage{Title: "Big", Type: "source", Content: "x\n"})(ctx, s, u) + }, + Schema: "# Schema", + }, + } + + first := processDir(context.Background(), cfg, "2026-10-20") + require.NotEmpty(t, first, "the refusal is reported") + assert.Contains(t, first[0].Error(), "too large") + second := processDir(context.Background(), cfg, "2026-10-20") + assert.Empty(t, second, "and reported once, not every poll") + + assert.Zero(t, calls.Load(), "no extraction") + _, err := os.Stat(big) + assert.NoError(t, err, "the note stays in raw/") + entries, _ := os.ReadDir(filepath.Join(brainDir, "raw", "processed")) + assert.Empty(t, entries, "nothing was archived") +} + +// hyperguild#199: a name near NAME_MAX cannot take a ".processed" suffix. Under +// inbox zero the marker is only the fallback when removal fails, and any failure +// there is remembered, so the note is still not handled every poll. +func TestProcessDir_ALongNameWhoseMarkerCannotBeWrittenIsNotHandledEveryPoll(t *testing.T) { + brainDir := setupBrainDir(t) + rawDir := filepath.Join(brainDir, "raw") + name := strings.Repeat("a", 250) + ".md" // 253 bytes; +".processed" exceeds 255 + require.NoError(t, os.WriteFile(filepath.Join(rawDir, name), []byte("Content."), 0o644)) + require.NoError(t, os.MkdirAll(filepath.Join(rawDir, "processed"), 0o755)) + require.NoError(t, os.Chmod(rawDir, 0o555)) // removal fails, so the marker fallback runs + t.Cleanup(func() { _ = os.Chmod(rawDir, 0o755) }) + cfg := Config{BrainDir: brainDir, Pipeline: pipeline.Config{ + Complete: successComplete(pipeline.RawPage{Title: "Long", Type: "source", Content: "x\n"}), Schema: "# Schema"}} + + for i := 0; i < 3; i++ { + errs := processDir(context.Background(), cfg, "2026-10-20") + // Without the remember step the second poll tries to archive a second + // copy under a name that is too long and errors, every poll. + assert.Empty(t, errs, "poll %d handled the note again", i) + } + archived, err := filepath.Glob(filepath.Join(rawDir, "processed", "2026-10-20", "*")) + require.NoError(t, err) + assert.Len(t, archived, 1) +} From 1025d777954e945de0a495bc0e20bd299a2dad2f Mon Sep 17 00:00:00 2001 From: Mathias Date: Tue, 20 Oct 2026 20:13:18 +0200 Subject: [PATCH 5/6] fix(watcher): unique archive copies, read caps at the read, stale markers (review of #210) From the independent review of #210: - F1: processed/ and failed/ copies are created O_EXCL with -2,-3 suffixes via createUnique; same-named notes from different folders no longer replace one another's only surviving copy - F2: extract.TextInMax / readAllMax cap the bytes actually read (a note can grow after the Lstat); extract.TextIn gets a 64 MiB default cap - F3: a regular marker older than its note describes an earlier note of the same name; it no longer hides the new one (warn + remove); non-regular markers still count as done (no writes through links) - F4: the planted-marker-link test is renamed to what it observes Each has a test that fails without it; the read-cap test uses an endless reader that errors at 8 MiB so a regression fails instead of exhausting memory. Co-Authored-By: Claude Sonnet 5.5 --- ingestion/internal/extract/extract.go | 25 +++- .../internal/extract/extract_cap_test.go | 62 +++++++++ ingestion/internal/watcher/park.go | 11 +- ingestion/internal/watcher/watcher.go | 112 ++++++++++++----- ingestion/internal/watcher/watcher_test.go | 119 +++++++++++++++++- .../internal/watcher/writes_redirect_test.go | 11 +- 6 files changed, 298 insertions(+), 42 deletions(-) create mode 100644 ingestion/internal/extract/extract_cap_test.go diff --git a/ingestion/internal/extract/extract.go b/ingestion/internal/extract/extract.go index eac7cea..1a96bed 100644 --- a/ingestion/internal/extract/extract.go +++ b/ingestion/internal/extract/extract.go @@ -2,6 +2,7 @@ package extract import ( + "errors" "fmt" "io" "os" @@ -50,6 +51,18 @@ func fileExt(path string) string { // external tool (pdf, docmark) are copied to a temp file from the handle // opened through the root, then extracted from the copy. func TextIn(r *os.Root, rel string) (string, error) { + return TextInMax(r, rel, DefaultMaxBytes) +} + +// DefaultMaxBytes bounds what TextIn reads or copies from one file. +const DefaultMaxBytes = 64 << 20 + +// ErrTooLarge reports a file that exceeded the read cap while being read: the +// size seen at an earlier Lstat is not a bound, the file may have grown. +var ErrTooLarge = errors.New("file exceeds the read cap") + +// TextInMax is TextIn with an explicit cap on the bytes read (hyperguild#198). +func TextInMax(r *os.Root, rel string, max int64) (string, error) { fi, err := r.Lstat(rel) if err != nil { return "", fmt.Errorf("stat %s: %w", rel, err) @@ -65,10 +78,13 @@ func TextIn(r *os.Root, rel string) (string, error) { ext := strings.ToLower(fileExt(rel)) switch ext { case ".md", ".txt": - b, err := io.ReadAll(f) + b, err := io.ReadAll(io.LimitReader(f, max+1)) if err != nil { return "", fmt.Errorf("read %s: %w", rel, err) } + if int64(len(b)) > max { + return "", fmt.Errorf("%s: %w (%d bytes)", rel, ErrTooLarge, max) + } return string(b), nil case ".pdf", ".docx", ".xlsx", ".pptx", ".png", ".jpg", ".jpeg": tmp, err := os.CreateTemp("", "ingest-*"+ext) @@ -76,10 +92,15 @@ func TextIn(r *os.Root, rel string) (string, error) { return "", fmt.Errorf("temp copy: %w", err) } defer func() { _ = os.Remove(tmp.Name()) }() - if _, err := io.Copy(tmp, f); err != nil { + n, err := io.Copy(tmp, io.LimitReader(f, max+1)) + if err != nil { _ = tmp.Close() return "", fmt.Errorf("copy %s: %w", rel, err) } + if n > max { + _ = tmp.Close() + return "", fmt.Errorf("%s: %w (%d bytes)", rel, ErrTooLarge, max) + } if err := tmp.Close(); err != nil { return "", fmt.Errorf("copy %s: %w", rel, err) } diff --git a/ingestion/internal/extract/extract_cap_test.go b/ingestion/internal/extract/extract_cap_test.go new file mode 100644 index 0000000..625aaa2 --- /dev/null +++ b/ingestion/internal/extract/extract_cap_test.go @@ -0,0 +1,62 @@ +package extract + +import ( + "errors" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func rootWith(t *testing.T, name string, size int) *os.Root { + t.Helper() + dir := t.TempDir() + require.NoError(t, os.WriteFile(filepath.Join(dir, name), []byte(strings.Repeat("x", size)), 0o644)) + r, err := os.OpenRoot(dir) + require.NoError(t, err) + t.Cleanup(func() { _ = r.Close() }) + return r +} + +// hyperguild#198: the size check in the watcher is a pre-check on Lstat; a note +// that grows before it is read must still be capped at the read. +func TestTextInMax(t *testing.T) { + tests := []struct { + name string + size int + max int64 + wantErr bool + }{ + {"under the cap", 100, 1024, false}, + {"exactly at the cap", 1024, 1024, false}, + {"one byte over", 1025, 1024, true}, + {"far over", 1 << 20, 1024, true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + r := rootWith(t, "n.md", tt.size) + got, err := TextInMax(r, "n.md", tt.max) + if tt.wantErr { + require.Error(t, err) + assert.True(t, errors.Is(err, ErrTooLarge)) + return + } + require.NoError(t, err) + assert.Len(t, got, tt.size) + }) + } +} + +func TestTextInMax_TheBinaryTempCopyIsCappedToo(t *testing.T) { + r := rootWith(t, "n.pdf", 4096) + _, err := TextInMax(r, "n.pdf", 1024) + require.Error(t, err) + assert.True(t, errors.Is(err, ErrTooLarge), "the temp copy of a pdf must not exceed the cap either: %v", err) +} + +func TestTextIn_HasADefaultCap(t *testing.T) { + assert.Positive(t, DefaultMaxBytes) +} diff --git a/ingestion/internal/watcher/park.go b/ingestion/internal/watcher/park.go index e0b1afb..36718e2 100644 --- a/ingestion/internal/watcher/park.go +++ b/ingestion/internal/watcher/park.go @@ -20,6 +20,13 @@ import ( // root, never replacing a parked note: on a name clash it tries -2, -3, ... // A needs-review that is a symlink, or any non-directory, is refused. func createParked(brainDir, filename string, data []byte) error { + return createUnique(brainDir, "raw/needs-review", filename, data) +} + +// createUnique writes data to dirRel/filename through the brain root, never +// replacing an existing file: on a name clash it tries -2, -3, ... A dirRel that +// is a symlink, or any non-directory, is refused. +func createUnique(brainDir, dirRel, filename string, data []byte) error { bfs, err := brainfs.Open(brainDir) if err != nil { return err @@ -32,7 +39,7 @@ func createParked(brainDir, filename string, data []byte) error { if n > 1 { name = fmt.Sprintf("%s-%d%s", stem, n, ext) } - err := bfs.CreateRegular("raw/needs-review/"+name, data) + err := bfs.CreateRegular(dirRel+"/"+name, data) if err == nil { return nil } @@ -40,5 +47,5 @@ func createParked(brainDir, filename string, data []byte) error { return err } } - return fmt.Errorf("no free name for %q in raw/needs-review", filename) + return fmt.Errorf("no free name for %q in %s", filename, dirRel) } diff --git a/ingestion/internal/watcher/watcher.go b/ingestion/internal/watcher/watcher.go index 14c6280..5837566 100644 --- a/ingestion/internal/watcher/watcher.go +++ b/ingestion/internal/watcher/watcher.go @@ -43,7 +43,7 @@ const defaultMaxNoteBytes = 64 << 20 // errTooLarge marks a note refused for its size; like a refused write it is // reported once and not retried until the note changes. -var errTooLarge = errors.New("note too large") +var errTooLarge = extract.ErrTooLarge // Start launches the watcher in a background goroutine. // It returns immediately. The watcher stops when ctx is cancelled. @@ -134,11 +134,11 @@ func processDir(ctx context.Context, cfg Config, date string) []error { // Any entry at the marker path counts (Lstat, not Stat): a dangling // marker symlink made os.Stat report nothing and the note was // reprocessed every interval (hyperguild#158). - if _, err := os.Lstat(path + ".processed"); err == nil { - return nil - } - if _, err := os.Lstat(path + ".failed"); err == nil { + if skip, why := markerBlocks(path); skip { return nil + } else if why != "" { + slog.Warn("watcher: ignoring a stale marker older than its note", "marker", why) + _ = root.Remove(filepath.ToSlash(relOf(rawDir, path)) + staleSuffix(why)) } // A note whose archive copy or marker was refused is not retried // until it changes: a refusal has no marker to stop the next round, @@ -188,7 +188,7 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d if before.Size() > limit { return fmt.Errorf("%s is %d bytes, over the %d byte cap: %w", filename, before.Size(), limit, errTooLarge) } - content, err := extract.TextIn(root, rel) + content, err := extract.TextInMax(root, rel, limit) if err != nil { return fmt.Errorf("extract text: %w", err) } @@ -206,7 +206,7 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d if runErr != nil { // Copy to failed/ and leave a .failed marker so we don't retry. Both // written through the brain root (hyperguild#158). - if cpErr := copyFromRoot(root, rel, cfg.BrainDir, "raw/failed/"+filename); cpErr != nil { + if cpErr := copyFromRoot(root, rel, cfg.BrainDir, "raw/failed", filename, limit); cpErr != nil { return fmt.Errorf("copy to failed: %w", cpErr) } if changedSince(root, rel, before) { @@ -226,14 +226,7 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d // Copy to processed/YYYY-MM-DD/ and leave a .processed marker so we don't re-ingest. // Written through the brain root (hyperguild#158). - dest := "raw/processed/" + date + "/" + filename - if _, err := os.Lstat(filepath.Join(cfg.BrainDir, filepath.FromSlash(dest))); err == nil { - // Archive copy already exists; append timestamp to avoid overwriting. - ext := filepath.Ext(filename) - base := strings.TrimSuffix(filename, ext) - dest = "raw/processed/" + date + "/" + base + "-" + time.Now().UTC().Format("150405") + ext - } - if err := copyFromRoot(root, rel, cfg.BrainDir, dest); err != nil { + if err := copyFromRoot(root, rel, cfg.BrainDir, "raw/processed/"+date, filename, limit); err != nil { return fmt.Errorf("copy to processed: %w", err) } if changedSince(root, rel, before) { @@ -337,27 +330,51 @@ func putBrainFile(brainDir, rel string, data []byte) error { return brainfs.PutFile(brainDir, rel, data) } -// copyFromRoot archives a raw/ file read through the root, refusing anything -// that is not a regular file, so a swapped-in symlink cannot copy an outside -// file into raw/processed/ or raw/failed/ (hyperguild#153). -func copyFromRoot(root *os.Root, rel, brainDir, dstRel string) error { - fi, err := root.Lstat(rel) +// copyFromRoot archives a raw/ file read through the root into dstDir under a +// name no existing file has (so a note never replaces another's copy: the copy +// is the only one left once the original is retired), refusing anything that is +// not a regular file and anything over max bytes at the read, not just at the +// earlier Lstat (hyperguild#158, #198). +func copyFromRoot(root *os.Root, rel, brainDir, dstDir, filename string, max int64) error { + data, err := readNoteMax(root, rel, max) if err != nil { return err } + return createUnique(brainDir, dstDir, filename, data) +} + +// readNoteMax reads a raw/ note through the root, at most max bytes. +func readNoteMax(root *os.Root, rel string, max int64) ([]byte, error) { + fi, err := root.Lstat(rel) + if err != nil { + return nil, err + } if !fi.Mode().IsRegular() { - return fmt.Errorf("%s: not a regular file", rel) + return nil, fmt.Errorf("%s: not a regular file", rel) } in, _, err := extract.OpenRegular(root, rel) if err != nil { - return err + return nil, err } defer func() { _ = in.Close() }() - data, err := io.ReadAll(in) - if err != nil { - return err + data, err := readAllMax(in, max) + if errors.Is(err, errTooLarge) { + return nil, fmt.Errorf("%s: %w (%d bytes)", rel, errTooLarge, max) } - return putBrainFile(brainDir, dstRel, data) + return data, err +} + +// readAllMax reads r but never more than max+1 bytes, so an oversize or growing +// file costs at most that much memory, and reports errTooLarge past max. +func readAllMax(r io.Reader, max int64) ([]byte, error) { + data, err := io.ReadAll(io.LimitReader(r, max+1)) + if err != nil { + return nil, err + } + if int64(len(data)) > max { + return nil, errTooLarge + } + return data, nil } // topDir is the first directory of a raw/-relative path, "" for a top-level file. @@ -397,12 +414,8 @@ func parkForReview(cfg Config, root *os.Root, rel, path, filename, content strin var data []byte if strings.ToLower(filepath.Ext(filename)) == ".pdf" { // Binary: park the bytes untouched; the proposal is in log.md only. - in, _, err := extract.OpenRegular(root, rel) - if err != nil { - return fmt.Errorf("park pdf: %w", err) - } - data, err = io.ReadAll(in) - _ = in.Close() + var err error + data, err = readNoteMax(root, rel, defaultMaxNoteBytes) if err != nil { return fmt.Errorf("park pdf: %w", err) } @@ -472,3 +485,38 @@ func cutFence(s string) (block, rest string, found bool) { } return "", "", false } + +// markerBlocks reports whether a marker next to path means "already handled". +// Any non-regular entry at the marker path counts (a dangling symlink must not +// be written through, hyperguild#158). A regular marker older than the note +// describes an EARLIER note of the same name, so it does not block this one: +// stale is the marker's path and the caller warns and removes it. +func markerBlocks(path string) (skip bool, stale string) { + note, err := os.Lstat(path) + if err != nil { + return false, "" + } + for _, suffix := range []string{".processed", ".failed"} { + m, err := os.Lstat(path + suffix) + if err != nil { + continue + } + if !m.Mode().IsRegular() || !m.ModTime().Before(note.ModTime()) { + return true, "" + } + stale = path + suffix + } + return false, stale +} + +func relOf(rawDir, path string) string { + rel, err := filepath.Rel(rawDir, path) + if err != nil { + return filepath.Base(path) + } + return rel +} + +func staleSuffix(markerPath string) string { + return filepath.Ext(markerPath) +} diff --git a/ingestion/internal/watcher/watcher_test.go b/ingestion/internal/watcher/watcher_test.go index 179d70c..3a5726d 100644 --- a/ingestion/internal/watcher/watcher_test.go +++ b/ingestion/internal/watcher/watcher_test.go @@ -4,6 +4,7 @@ package watcher import ( "context" "encoding/json" + "errors" "fmt" "os" "path/filepath" @@ -497,7 +498,7 @@ func TestProcessDir_ANoteOverTheSizeCapIsRefusedOnceAndKept(t *testing.T) { first := processDir(context.Background(), cfg, "2026-10-20") require.NotEmpty(t, first, "the refusal is reported") - assert.Contains(t, first[0].Error(), "too large") + assert.Contains(t, first[0].Error(), "over the 1024 byte cap") second := processDir(context.Background(), cfg, "2026-10-20") assert.Empty(t, second, "and reported once, not every poll") @@ -532,3 +533,119 @@ func TestProcessDir_ALongNameWhoseMarkerCannotBeWrittenIsNotHandledEveryPoll(t * require.NoError(t, err) assert.Len(t, archived, 1) } + +// Three notes named n.md in three folders, one poll. The archive and failed/ +// copies are the only copy once the originals are retired, so none may replace +// another. +func TestProcessDir_SameNamedNotesAreAllArchivedNoneOverwritten(t *testing.T) { + for _, outcome := range []string{"processed", "failed"} { + t.Run(outcome, func(t *testing.T) { + brainDir := setupBrainDir(t) + for _, d := range []string{"a", "b", "c"} { + require.NoError(t, os.MkdirAll(filepath.Join(brainDir, "raw", d), 0o755)) + require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", d, "n.md"), []byte("CONTENT-"+d), 0o644)) + } + cfg := Config{BrainDir: brainDir, Pipeline: pipeline.Config{Schema: "# Schema", + Complete: successComplete(pipeline.RawPage{Title: "N", Type: "source", Content: "x\n"})}} + if outcome == "failed" { + cfg.Pipeline.Complete = errorComplete + } + processDir(context.Background(), cfg, "2026-10-20") + + var all string + files, _ := filepath.Glob(filepath.Join(brainDir, "raw", outcome, "*", "*")) + more, _ := filepath.Glob(filepath.Join(brainDir, "raw", outcome, "*")) + for _, f := range append(files, more...) { + if b, err := os.ReadFile(f); err == nil { + all += string(b) + } + } + for _, d := range []string{"a", "b", "c"} { + assert.Contains(t, all, "CONTENT-"+d, "the copy of note %s was overwritten", d) + } + }) + } +} + +// A marker older than the note describes an earlier note of the same name; it +// must not hide the new one, silently, forever (the legacy .processed markers +// from before inbox zero are exactly this). +func TestProcessDir_AStaleMarkerDoesNotHideANewerNoteOfTheSameName(t *testing.T) { + brainDir := setupBrainDir(t) + rawDir := filepath.Join(brainDir, "raw") + marker := filepath.Join(rawDir, "idea.md.processed") + require.NoError(t, os.WriteFile(marker, []byte("2026-06-16"), 0o644)) + old := time.Now().Add(-48 * time.Hour) + require.NoError(t, os.Chtimes(marker, old, old)) + require.NoError(t, os.WriteFile(filepath.Join(rawDir, "idea.md"), []byte("A new idea, same file name."), 0o644)) + + var calls atomic.Int32 + cfg := Config{BrainDir: brainDir, Pipeline: pipeline.Config{Schema: "# Schema", + Complete: func(ctx context.Context, s, u string) (string, error) { + calls.Add(1) + return successComplete(pipeline.RawPage{Title: "Idea", Type: "source", Content: "x\n"})(ctx, s, u) + }}} + require.Empty(t, processDir(context.Background(), cfg, "2026-10-20")) + + assert.Equal(t, int32(1), calls.Load(), "the new note is processed") + _, err := os.Stat(filepath.Join(rawDir, "idea.md")) + assert.ErrorIs(t, err, os.ErrNotExist, "and retired") + _, err = os.Lstat(marker) + assert.ErrorIs(t, err, os.ErrNotExist, "the stale marker is cleaned up") +} + +func TestProcessDir_AMarkerNewerThanTheNoteStillMeansDone(t *testing.T) { + brainDir := setupBrainDir(t) + rawDir := filepath.Join(brainDir, "raw") + note := filepath.Join(rawDir, "stuck.md") + require.NoError(t, os.WriteFile(note, []byte("Content."), 0o644)) + past := time.Now().Add(-time.Hour) + require.NoError(t, os.Chtimes(note, past, past)) + require.NoError(t, os.WriteFile(note+".processed", []byte("d"), 0o644)) // newer than the note + + var calls atomic.Int32 + cfg := Config{BrainDir: brainDir, Pipeline: pipeline.Config{Schema: "# Schema", + Complete: func(context.Context, string, string) (string, error) { calls.Add(1); return "[]", nil }}} + processDir(context.Background(), cfg, "2026-10-20") + assert.Zero(t, calls.Load(), "a fallback marker written after the note still stops reprocessing") +} + +// The archive copy is a second read of the note; the cap must hold there too, +// not only at the earlier Lstat (a note can grow in between). +func TestReadNoteMax(t *testing.T) { + dir := t.TempDir() + require.NoError(t, os.WriteFile(filepath.Join(dir, "n.md"), []byte(strings.Repeat("x", 100)), 0o644)) + root, err := os.OpenRoot(dir) + require.NoError(t, err) + defer func() { _ = root.Close() }() + + got, err := readNoteMax(root, "n.md", 100) + require.NoError(t, err) + assert.Len(t, got, 100, "exactly at the cap is allowed") + + _, err = readNoteMax(root, "n.md", 99) + require.Error(t, err) + assert.ErrorIs(t, err, errTooLarge) +} + +// countingReader reports how much was pulled from it: an endless file must be +// abandoned at max+1 bytes, not slurped. +type countingReader struct{ n int64 } + +func (c *countingReader) Read(p []byte) (int, error) { + if c.n > 8<<20 { // a regression must fail the test, not exhaust memory + return 0, errors.New("endless reader drained") + } + for i := range p { + p[i] = 'x' + } + c.n += int64(len(p)) + return len(p), nil +} + +func TestReadAllMax_StopsReadingAtTheCap(t *testing.T) { + c := &countingReader{} + _, err := readAllMax(c, 1000) + require.ErrorIs(t, err, errTooLarge) + assert.LessOrEqual(t, c.n, int64(1000+1+32*1024), "pulled far more than the cap from an endless reader: %d", c.n) +} diff --git a/ingestion/internal/watcher/writes_redirect_test.go b/ingestion/internal/watcher/writes_redirect_test.go index 9649a0c..c9560e2 100644 --- a/ingestion/internal/watcher/writes_redirect_test.go +++ b/ingestion/internal/watcher/writes_redirect_test.go @@ -103,11 +103,12 @@ func TestProcessDirTreatsAnyEntryAtTheMarkerPathAsAMarker(t *testing.T) { } // A link planted at the marker path while pipeline.Run is in flight (in -// production, an LLM call) is never written through, and the note is processed -// once: under inbox zero the original is removed, so there is no marker write -// left to redirect, and removing the link (as symlinkguard does) does not make -// the watcher call the model again. -func TestProcessDirNeverWritesThroughAMarkerLinkPlantedDuringTheRun(t *testing.T) { +// production, an LLM call) ends up with its target untouched, and the note is +// processed once, even after the link is removed (as symlinkguard does). Under +// inbox zero the original is removed, so the marker write that this scenario +// was written to redirect is never reached: this test does NOT pin that the +// marker-fallback write goes through the brain root (see the PR description). +func TestProcessDirLeavesAMarkerLinkPlantedDuringTheRunUntouchedAndProcessesTheNoteOnce(t *testing.T) { for _, kind := range []string{"processed", "failed"} { for _, target := range []string{"outside", "in-brain"} { t.Run(kind+"/"+target, func(t *testing.T) { From 6717221989421135a2765a7be6b6a13aa9ea825f Mon Sep 17 00:00:00 2001 From: Mathias Date: Tue, 20 Oct 2026 20:17:10 +0200 Subject: [PATCH 6/6] fix(watcher): a held note is reported once; future-dated and both stale markers handled (review of #210, pass 2) - a marker that blocks a note is warned about once, not silently forever - a marker dated more than 10 minutes ahead (clock skew, a peer) is not believed and is treated as stale - every stale marker is removed, not only the last; removal errors are logged Each has a test that fails without it. Co-Authored-By: Claude Sonnet 5.5 --- ingestion/internal/watcher/watcher.go | 53 +++++++++++++------ ingestion/internal/watcher/watcher_test.go | 61 ++++++++++++++++++++++ 2 files changed, 98 insertions(+), 16 deletions(-) diff --git a/ingestion/internal/watcher/watcher.go b/ingestion/internal/watcher/watcher.go index 5837566..ca742c4 100644 --- a/ingestion/internal/watcher/watcher.go +++ b/ingestion/internal/watcher/watcher.go @@ -134,11 +134,16 @@ func processDir(ctx context.Context, cfg Config, date string) []error { // Any entry at the marker path counts (Lstat, not Stat): a dangling // marker symlink made os.Stat report nothing and the note was // reprocessed every interval (hyperguild#158). - if skip, why := markerBlocks(path); skip { + if skip, stale := markerBlocks(path); skip { + warnMarkerOnce(path) return nil - } else if why != "" { - slog.Warn("watcher: ignoring a stale marker older than its note", "marker", why) - _ = root.Remove(filepath.ToSlash(relOf(rawDir, path)) + staleSuffix(why)) + } else { + for _, m := range stale { + slog.Warn("watcher: ignoring a stale marker older than its note", "marker", m) + if rmErr := root.Remove(relOf(rawDir, m)); rmErr != nil { + slog.Error("watcher: failed to remove a stale marker", "marker", m, "error", rmErr) + } + } } // A note whose archive copy or marker was refused is not retried // until it changes: a refusal has no marker to stop the next round, @@ -488,27 +493,47 @@ func cutFence(s string) (block, rest string, found bool) { // markerBlocks reports whether a marker next to path means "already handled". // Any non-regular entry at the marker path counts (a dangling symlink must not -// be written through, hyperguild#158). A regular marker older than the note -// describes an EARLIER note of the same name, so it does not block this one: -// stale is the marker's path and the caller warns and removes it. -func markerBlocks(path string) (skip bool, stale string) { +// be written through, hyperguild#158). A regular marker describes an EARLIER +// note of the same name, so it does not block this one, and is returned in +// stale, when it is older than the note or dated implausibly far in the future +// (clock skew, a peer): those are removed by the caller. Heuristic by nature: +// a marker at least as new as the note still means "done". +func markerBlocks(path string) (skip bool, stale []string) { note, err := os.Lstat(path) if err != nil { - return false, "" + return false, nil } for _, suffix := range []string{".processed", ".failed"} { m, err := os.Lstat(path + suffix) if err != nil { continue } - if !m.Mode().IsRegular() || !m.ModTime().Before(note.ModTime()) { - return true, "" + if !m.Mode().IsRegular() { + return true, nil } - stale = path + suffix + if m.ModTime().Before(note.ModTime()) || m.ModTime().After(time.Now().Add(markerSkew)) { + stale = append(stale, path+suffix) + continue + } + return true, nil } return false, stale } +// markerSkew is how far ahead of now a marker may be dated before it is not +// believed. +const markerSkew = 10 * time.Minute + +// markerWarned remembers notes already reported as held by a marker, so the +// 30-second poll says it once instead of forever. +var markerWarned sync.Map + +func warnMarkerOnce(path string) { + if _, seen := markerWarned.LoadOrStore(path, struct{}{}); !seen { + slog.Warn("watcher: note is held by a marker newer than itself and will not be processed; delete the marker or edit the note", "file", path) + } +} + func relOf(rawDir, path string) string { rel, err := filepath.Rel(rawDir, path) if err != nil { @@ -516,7 +541,3 @@ func relOf(rawDir, path string) string { } return rel } - -func staleSuffix(markerPath string) string { - return filepath.Ext(markerPath) -} diff --git a/ingestion/internal/watcher/watcher_test.go b/ingestion/internal/watcher/watcher_test.go index 3a5726d..d29f794 100644 --- a/ingestion/internal/watcher/watcher_test.go +++ b/ingestion/internal/watcher/watcher_test.go @@ -6,6 +6,7 @@ import ( "encoding/json" "errors" "fmt" + "log/slog" "os" "path/filepath" "strings" @@ -649,3 +650,63 @@ func TestReadAllMax_StopsReadingAtTheCap(t *testing.T) { require.ErrorIs(t, err, errTooLarge) assert.LessOrEqual(t, c.n, int64(1000+1+32*1024), "pulled far more than the cap from an endless reader: %d", c.n) } + +// From the second review of #210: a marker dated in the future (clock skew, +// Syncthing) must not hide the note for good, and when a marker does block it +// has to say so, once. +func TestProcessDir_AFutureDatedMarkerDoesNotHideTheNote(t *testing.T) { + brainDir := setupBrainDir(t) + rawDir := filepath.Join(brainDir, "raw") + marker := filepath.Join(rawDir, "idea.md.processed") + require.NoError(t, os.WriteFile(marker, []byte("x"), 0o644)) + future := time.Now().Add(72 * time.Hour) + require.NoError(t, os.Chtimes(marker, future, future)) + require.NoError(t, os.WriteFile(filepath.Join(rawDir, "idea.md"), []byte("A new idea."), 0o644)) + + var calls atomic.Int32 + cfg := Config{BrainDir: brainDir, Pipeline: pipeline.Config{Schema: "# Schema", + Complete: func(ctx context.Context, s, u string) (string, error) { + calls.Add(1) + return successComplete(pipeline.RawPage{Title: "Idea", Type: "source", Content: "x\n"})(ctx, s, u) + }}} + processDir(context.Background(), cfg, "2026-10-20") + assert.Equal(t, int32(1), calls.Load(), "a marker from the future is not evidence the note was handled") +} + +func TestProcessDir_BothStaleMarkersAreRemoved(t *testing.T) { + brainDir := setupBrainDir(t) + rawDir := filepath.Join(brainDir, "raw") + old := time.Now().Add(-48 * time.Hour) + for _, suffix := range []string{".processed", ".failed"} { + require.NoError(t, os.WriteFile(filepath.Join(rawDir, "idea.md"+suffix), []byte("x"), 0o644)) + require.NoError(t, os.Chtimes(filepath.Join(rawDir, "idea.md"+suffix), old, old)) + } + require.NoError(t, os.WriteFile(filepath.Join(rawDir, "idea.md"), []byte("A new idea."), 0o644)) + cfg := Config{BrainDir: brainDir, Pipeline: pipeline.Config{Schema: "# Schema", + Complete: successComplete(pipeline.RawPage{Title: "Idea", Type: "source", Content: "x\n"})}} + processDir(context.Background(), cfg, "2026-10-20") + for _, suffix := range []string{".processed", ".failed"} { + _, err := os.Lstat(filepath.Join(rawDir, "idea.md"+suffix)) + assert.ErrorIs(t, err, os.ErrNotExist, "orphan %s marker left behind", suffix) + } +} + +func TestProcessDir_AMarkerThatBlocksANoteIsReportedOnce(t *testing.T) { + var logs strings.Builder + prev := slog.Default() + slog.SetDefault(slog.New(slog.NewTextHandler(&logs, nil))) + t.Cleanup(func() { slog.SetDefault(prev) }) + + brainDir := setupBrainDir(t) + rawDir := filepath.Join(brainDir, "raw") + note := filepath.Join(rawDir, "held.md") + require.NoError(t, os.WriteFile(note, []byte("Content."), 0o644)) + past := time.Now().Add(-time.Hour) + require.NoError(t, os.Chtimes(note, past, past)) + require.NoError(t, os.WriteFile(note+".processed", []byte("d"), 0o644)) // newer: blocks + cfg := Config{BrainDir: brainDir, Pipeline: pipeline.Config{Schema: "# Schema", Complete: errorComplete}} + for i := 0; i < 3; i++ { + processDir(context.Background(), cfg, "2026-10-20") + } + assert.Equal(t, 1, strings.Count(logs.String(), "held by a marker"), "said once, not every poll: %s", logs.String()) +}