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

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