fix(watcher): keep a note that changed while it was being ingested
The archive holds the content read at the start; if Syncthing replaced the note during the (minutes-long) LLM pass, removing the original would drop the newer content. Compare size+mtime before removing; leave it for the next poll. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -156,6 +156,10 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d
|
|||||||
return fmt.Errorf("relative path: %w", err)
|
return fmt.Errorf("relative path: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
before, err := root.Lstat(rel)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("stat before ingest: %w", err)
|
||||||
|
}
|
||||||
content, err := extract.TextIn(root, rel)
|
content, err := extract.TextIn(root, rel)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("extract text: %w", err)
|
return fmt.Errorf("extract text: %w", err)
|
||||||
@@ -172,6 +176,10 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d
|
|||||||
if cpErr := copyFromRoot(root, rel, dest); cpErr != nil {
|
if cpErr := copyFromRoot(root, rel, dest); cpErr != nil {
|
||||||
return fmt.Errorf("copy to failed: %w", cpErr)
|
return fmt.Errorf("copy to failed: %w", cpErr)
|
||||||
}
|
}
|
||||||
|
if changedSince(root, rel, before) {
|
||||||
|
slog.Warn("watcher: note changed during ingest; leaving it in raw/ for the next poll", "file", filename)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
retire(cfg, root, rel, path, ".failed", runErr.Error())
|
retire(cfg, root, rel, path, ".failed", runErr.Error())
|
||||||
|
|
||||||
slog.Warn("watcher: file failed", "file", filename, "error", runErr)
|
slog.Warn("watcher: file failed", "file", filename, "error", runErr)
|
||||||
@@ -198,12 +206,27 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d
|
|||||||
if err := copyFromRoot(root, rel, dest); err != nil {
|
if err := copyFromRoot(root, rel, dest); err != nil {
|
||||||
return fmt.Errorf("copy to processed: %w", err)
|
return fmt.Errorf("copy to processed: %w", err)
|
||||||
}
|
}
|
||||||
|
if changedSince(root, rel, before) {
|
||||||
|
slog.Warn("watcher: note changed during ingest; leaving it in raw/ for the next poll", "file", filename)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
retire(cfg, root, rel, path, ".processed", date)
|
retire(cfg, root, rel, path, ".processed", date)
|
||||||
|
|
||||||
slog.Info("watcher: file processed", "file", filename, "source", source)
|
slog.Info("watcher: file processed", "file", filename, "source", source)
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// changedSince reports whether the note at rel differs (size or mtime) from
|
||||||
|
// before. A note replaced mid-ingest (Syncthing) has newer content than the
|
||||||
|
// archive copy, so it must not be removed.
|
||||||
|
func changedSince(root *os.Root, rel string, before os.FileInfo) bool {
|
||||||
|
now, err := root.Lstat(rel)
|
||||||
|
if err != nil {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
return now.Size() != before.Size() || !now.ModTime().Equal(before.ModTime())
|
||||||
|
}
|
||||||
|
|
||||||
// retire removes an archived original from raw/ (inbox zero). If the removal
|
// retire removes an archived original from raw/ (inbox zero). If the removal
|
||||||
// fails it falls back to a marker file so the watcher still skips the file.
|
// fails it falls back to a marker file so the watcher still skips the file.
|
||||||
func retire(cfg Config, root *os.Root, rel, path, markerSuffix, markerBody string) {
|
func retire(cfg Config, root *os.Root, rel, path, markerSuffix, markerBody string) {
|
||||||
|
|||||||
@@ -444,3 +444,27 @@ func TestProcessDir_KeepsTheOriginalWhenTheArchiveCopyFails(t *testing.T) {
|
|||||||
_, err := os.Stat(note)
|
_, err := os.Stat(note)
|
||||||
assert.NoError(t, err, "original stays in raw/ when it could not be archived")
|
assert.NoError(t, err, "original stays in raw/ when it could not be archived")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// If the note is replaced (e.g. by Syncthing) while it is being ingested, the
|
||||||
|
// archive holds the OLD content; removing the original would lose the new one.
|
||||||
|
func TestProcessDir_KeepsANoteThatChangedWhileItWasBeingIngested(t *testing.T) {
|
||||||
|
brainDir := setupBrainDir(t)
|
||||||
|
note := filepath.Join(brainDir, "raw", "edited.md")
|
||||||
|
require.NoError(t, os.WriteFile(note, []byte("old"), 0o644))
|
||||||
|
|
||||||
|
cfg := Config{
|
||||||
|
BrainDir: brainDir,
|
||||||
|
Pipeline: pipeline.Config{
|
||||||
|
Complete: func(ctx context.Context, s, u string) (string, error) {
|
||||||
|
require.NoError(t, os.WriteFile(note, []byte("newer and longer content"), 0o644))
|
||||||
|
return successComplete(pipeline.RawPage{Title: "Edited", Type: "source", Subtype: "article", Content: "x\n"})(ctx, s, u)
|
||||||
|
},
|
||||||
|
Schema: "# Schema",
|
||||||
|
},
|
||||||
|
}
|
||||||
|
processDir(context.Background(), cfg, "2026-10-19")
|
||||||
|
|
||||||
|
got, err := os.ReadFile(note)
|
||||||
|
require.NoError(t, err, "a note that changed during ingest must stay in raw/")
|
||||||
|
assert.Equal(t, "newer and longer content", string(got))
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user