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