From 6717221989421135a2765a7be6b6a13aa9ea825f Mon Sep 17 00:00:00 2001 From: Mathias Date: Tue, 20 Oct 2026 20:17:10 +0200 Subject: [PATCH] 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()) +}