Merge branch 'wt/watcher-write-safety' into wt/web-clip

2026-10-20 20:18:39 +02:00
parent 8988c291ca 6717221989
commit 603d2f9dfb
6 changed files with 380 additions and 42 deletions
@@ -45,7 +45,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.
@@ -136,11 +136,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 _, err := os.Lstat(path + ".processed"); err == nil {
-			return nil
-		}
-		if _, err := os.Lstat(path + ".failed"); err == nil {
+		if skip, stale := markerBlocks(path); skip {
+			warnMarkerOnce(path)
 			return nil
+		} 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,
@@ -190,7 +195,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)
 	}
@@ -208,7 +213,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) {
@@ -228,14 +233,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) {
@@ -339,27 +337,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.
@@ -399,12 +421,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)
 		}
@@ -474,3 +492,54 @@ 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 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, nil
+	}
+	for _, suffix := range []string{".processed", ".failed"} {
+		m, err := os.Lstat(path + suffix)
+		if err != nil {
+			continue
+		}
+		if !m.Mode().IsRegular() {
+			return true, nil
+		}
+		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 {
+		return filepath.Base(path)
+	}
+	return rel
+}