fix: pipeline, backfill, watcher and archive writes go through the brain root (#158 part B) #195
@@ -17,6 +17,7 @@ import (
|
|||||||
|
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/api"
|
"github.com/mathiasbq/hyperguild/ingestion/internal/api"
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/audit"
|
"github.com/mathiasbq/hyperguild/ingestion/internal/audit"
|
||||||
|
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/capture"
|
"github.com/mathiasbq/hyperguild/ingestion/internal/capture"
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/capturehttp"
|
"github.com/mathiasbq/hyperguild/ingestion/internal/capturehttp"
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/classification"
|
"github.com/mathiasbq/hyperguild/ingestion/internal/classification"
|
||||||
@@ -89,11 +90,15 @@ func (s *claudeSink) Ingest(ctx context.Context, b claudewatcher.Batch) error {
|
|||||||
sb.WriteString("\n\n")
|
sb.WriteString("\n\n")
|
||||||
}
|
}
|
||||||
slug := "session-" + b.Host + "-" + b.SessionID
|
slug := "session-" + b.Host + "-" + b.SessionID
|
||||||
dest := filepath.Join(s.brainDir, "archive", "claude-sessions", b.Host, slug+".md")
|
dest := "archive/claude-sessions/" + b.Host + "/" + slug + ".md"
|
||||||
if err := os.MkdirAll(filepath.Dir(dest), 0o755); err != nil {
|
// Written through the brain root, never through a symlinked directory
|
||||||
return fmt.Errorf("create claude-sessions archive dir: %w", err)
|
// or file (hyperguild#158).
|
||||||
|
bfs, err := brainfs.Open(s.brainDir)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("open brain: %w", err)
|
||||||
}
|
}
|
||||||
if err := os.WriteFile(dest, []byte(sb.String()), 0o644); err != nil {
|
defer func() { _ = bfs.Close() }()
|
||||||
|
if err := bfs.PutRegular(dest, []byte(sb.String())); err != nil {
|
||||||
return fmt.Errorf("write claude session archive: %w", err)
|
return fmt.Errorf("write claude session archive: %w", err)
|
||||||
}
|
}
|
||||||
s.logger.Debug("claude session archived (non-indexed)", "path", dest)
|
s.logger.Debug("claude session archived (non-indexed)", "path", dest)
|
||||||
|
|||||||
@@ -243,6 +243,27 @@ func (b *Brain) RealParent(rel string) error {
|
|||||||
return b.realDir(path.Dir(rel))
|
return b.realDir(path.Dir(rel))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// PutFile opens the brain dir as a root and calls PutRegular, for one-off
|
||||||
|
// writers that hold no Brain.
|
||||||
|
func PutFile(brainDir, rel string, data []byte) error {
|
||||||
|
b, err := Open(brainDir)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
defer func() { _ = b.Close() }()
|
||||||
|
return b.PutRegular(rel, data)
|
||||||
|
}
|
||||||
|
|
||||||
|
// AppendFile opens the brain dir as a root and calls AppendRegular.
|
||||||
|
func AppendFile(brainDir, rel string, data []byte) error {
|
||||||
|
b, err := Open(brainDir)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
defer func() { _ = b.Close() }()
|
||||||
|
return b.AppendRegular(rel, data)
|
||||||
|
}
|
||||||
|
|
||||||
// mkdirReal makes sure every component of rel exists as a real directory,
|
// mkdirReal makes sure every component of rel exists as a real directory,
|
||||||
// creating missing ones through the root; an existing component that is
|
// creating missing ones through the root; an existing component that is
|
||||||
// not a real directory is ErrNotRegular.
|
// not a real directory is ErrNotRegular.
|
||||||
|
|||||||
@@ -7,8 +7,6 @@ import (
|
|||||||
"fmt"
|
"fmt"
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
|
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
|
||||||
"io/fs"
|
"io/fs"
|
||||||
"os"
|
|
||||||
"path/filepath"
|
|
||||||
"strings"
|
"strings"
|
||||||
|
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/wiki"
|
"github.com/mathiasbq/hyperguild/ingestion/internal/wiki"
|
||||||
@@ -90,8 +88,7 @@ func BackfillRefs(ctx context.Context, brainDir string) (int, error) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
for relPath, page := range pending {
|
for relPath, page := range pending {
|
||||||
dest := filepath.Join(brainDir, filepath.FromSlash(relPath))
|
if err := bfs.PutRegular(relPath, []byte(page.Content)); err != nil {
|
||||||
if err := os.WriteFile(dest, []byte(page.Content), 0o644); err != nil {
|
|
||||||
return 0, fmt.Errorf("write %s: %w", relPath, err)
|
return 0, fmt.Errorf("write %s: %w", relPath, err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -9,8 +9,6 @@ import (
|
|||||||
"fmt"
|
"fmt"
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
|
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
|
||||||
"io/fs"
|
"io/fs"
|
||||||
"os"
|
|
||||||
"path/filepath"
|
|
||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -128,10 +126,6 @@ func buildAndWrite(ctx context.Context, rawPages []RawPage, sourceSlug, date, br
|
|||||||
var written []string
|
var written []string
|
||||||
for _, page := range merged {
|
for _, page := range merged {
|
||||||
if !dryRun {
|
if !dryRun {
|
||||||
dest := filepath.Join(brainDir, filepath.FromSlash(page.Path))
|
|
||||||
if err := os.MkdirAll(filepath.Dir(dest), 0o755); err != nil {
|
|
||||||
return Result{}, fmt.Errorf("mkdir for %s: %w", page.Path, err)
|
|
||||||
}
|
|
||||||
content, conflicts, err := reviseOrCreate(brainDir, page)
|
content, conflicts, err := reviseOrCreate(brainDir, page)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return Result{}, err
|
return Result{}, err
|
||||||
@@ -150,7 +144,9 @@ func buildAndWrite(ctx context.Context, rawPages []RawPage, sourceSlug, date, br
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return Result{}, fmt.Errorf("stamp %s: %w", page.Path, err)
|
return Result{}, fmt.Errorf("stamp %s: %w", page.Path, err)
|
||||||
}
|
}
|
||||||
if err := os.WriteFile(dest, []byte(stamped), 0o644); err != nil {
|
// Written through the brain root, never through a symlinked
|
||||||
|
// directory or page (hyperguild#158).
|
||||||
|
if err := writeBrainFile(brainDir, page.Path, []byte(stamped)); err != nil {
|
||||||
return Result{}, fmt.Errorf("write %s: %w", page.Path, err)
|
return Result{}, fmt.Errorf("write %s: %w", page.Path, err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -201,6 +197,17 @@ func readSourcePage(brainDir, sourceSlug string) string {
|
|||||||
return string(raw)
|
return string(raw)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// writeBrainFile creates or replaces one brain-relative file through the
|
||||||
|
// brain root (hyperguild#158).
|
||||||
|
func writeBrainFile(brainDir, rel string, data []byte) error {
|
||||||
|
bfs, err := brainfs.Open(brainDir)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
defer func() { _ = bfs.Close() }()
|
||||||
|
return bfs.PutRegular(rel, data)
|
||||||
|
}
|
||||||
|
|
||||||
// readBrainFile reads one brain-relative file through the brain root,
|
// readBrainFile reads one brain-relative file through the brain root,
|
||||||
// regular files only (hyperguild#157).
|
// regular files only (hyperguild#157).
|
||||||
func readBrainFile(brainDir, rel string) ([]byte, error) {
|
func readBrainFile(brainDir, rel string) ([]byte, error) {
|
||||||
|
|||||||
@@ -165,3 +165,28 @@ func TestBackfillRefsDoesNotCopyAPageSwappedInBehindASymlink(t *testing.T) {
|
|||||||
require.Contains(t, read(), "[[s|S]]", "positive control: the regular page gets the source ref")
|
require.Contains(t, read(), "[[s|S]]", "positive control: the regular page gets the source ref")
|
||||||
brainfstest.AssertNoLeakFrom(t, page, inside, target, 3*time.Second, read)
|
brainfstest.AssertNoLeakFrom(t, page, inside, target, 3*time.Second, read)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// hyperguild#158: BackfillRefs read a regular concept page, a symlink could
|
||||||
|
// be swapped in, and os.WriteFile then followed it and overwrote the link's
|
||||||
|
// target. The target is not restored between rounds here, so any
|
||||||
|
// write-through shows.
|
||||||
|
func TestBackfillRefsNeverWritesThroughAPageSwappedForASymlink(t *testing.T) {
|
||||||
|
brain := t.TempDir()
|
||||||
|
mkdirs(t, brain, "wiki/sources", "wiki/concepts", "raw")
|
||||||
|
require.NoError(t, os.WriteFile(filepath.Join(brain, "wiki", "sources", "s.md"), []byte("---\ntitle: S\n---\n[[c|C]]\n"), 0o644))
|
||||||
|
page := filepath.Join(brain, "wiki", "concepts", "c.md")
|
||||||
|
target := filepath.Join(brain, "raw", "secret.md")
|
||||||
|
require.NoError(t, os.WriteFile(target, []byte("untouched"), 0o644))
|
||||||
|
inside := "---\ntitle: C\n---\nbody\n"
|
||||||
|
require.NoError(t, os.WriteFile(page, []byte(inside), 0o644))
|
||||||
|
brainfstest.StartSwap(t, page, inside, target, false)
|
||||||
|
ok := 0
|
||||||
|
for deadline := time.Now().Add(2 * time.Second); time.Now().Before(deadline); {
|
||||||
|
if _, err := BackfillRefs(context.Background(), brain); err == nil {
|
||||||
|
ok++
|
||||||
|
}
|
||||||
|
}
|
||||||
|
got, _ := os.ReadFile(target)
|
||||||
|
assert.Equal(t, "untouched", string(got), "BackfillRefs wrote through a swapped-in symlink")
|
||||||
|
assert.Positive(t, ok, "positive control: some backfills ran")
|
||||||
|
}
|
||||||
|
|||||||
@@ -12,6 +12,7 @@ import (
|
|||||||
"time"
|
"time"
|
||||||
"unicode"
|
"unicode"
|
||||||
|
|
||||||
|
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/extract"
|
"github.com/mathiasbq/hyperguild/ingestion/internal/extract"
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/pipeline"
|
"github.com/mathiasbq/hyperguild/ingestion/internal/pipeline"
|
||||||
)
|
)
|
||||||
@@ -145,16 +146,12 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d
|
|||||||
|
|
||||||
_, runErr := pipeline.Run(ctx, cfg.Pipeline, cfg.BrainDir, content, source, false)
|
_, runErr := pipeline.Run(ctx, cfg.Pipeline, cfg.BrainDir, content, source, false)
|
||||||
if runErr != nil {
|
if runErr != nil {
|
||||||
// Copy to failed/ and leave a .failed marker so we don't retry.
|
// Copy to failed/ and leave a .failed marker so we don't retry. Both
|
||||||
failedDir := filepath.Join(cfg.BrainDir, "raw", "failed")
|
// written through the brain root (hyperguild#158).
|
||||||
if mkErr := os.MkdirAll(failedDir, 0o755); mkErr != nil {
|
if cpErr := copyFromRoot(root, rel, cfg.BrainDir, "raw/failed/"+filename); cpErr != nil {
|
||||||
return fmt.Errorf("mkdir failed dir: %w", mkErr)
|
|
||||||
}
|
|
||||||
dest := filepath.Join(failedDir, filename)
|
|
||||||
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 mkErr := os.WriteFile(path+".failed", []byte(runErr.Error()), 0o644); mkErr != nil {
|
if mkErr := putBrainFile(cfg.BrainDir, "raw/"+filepath.ToSlash(rel)+".failed", []byte(runErr.Error())); mkErr != nil {
|
||||||
slog.Error("watcher: failed to write .failed marker", "error", mkErr)
|
slog.Error("watcher: failed to write .failed marker", "error", mkErr)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -168,21 +165,18 @@ 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.
|
// Copy to processed/YYYY-MM-DD/ and leave a .processed marker so we don't re-ingest.
|
||||||
processedDir := filepath.Join(cfg.BrainDir, "raw", "processed", date)
|
// Written through the brain root (hyperguild#158).
|
||||||
if err := os.MkdirAll(processedDir, 0o755); err != nil {
|
dest := "raw/processed/" + date + "/" + filename
|
||||||
return fmt.Errorf("mkdir processed dir: %w", err)
|
if _, err := os.Lstat(filepath.Join(cfg.BrainDir, filepath.FromSlash(dest))); err == nil {
|
||||||
}
|
|
||||||
dest := filepath.Join(processedDir, filename)
|
|
||||||
if _, err := os.Stat(dest); err == nil {
|
|
||||||
// Archive copy already exists; append timestamp to avoid overwriting.
|
// Archive copy already exists; append timestamp to avoid overwriting.
|
||||||
ext := filepath.Ext(filename)
|
ext := filepath.Ext(filename)
|
||||||
base := strings.TrimSuffix(filename, ext)
|
base := strings.TrimSuffix(filename, ext)
|
||||||
dest = filepath.Join(processedDir, base+"-"+time.Now().UTC().Format("150405")+ext)
|
dest = "raw/processed/" + date + "/" + base + "-" + time.Now().UTC().Format("150405") + ext
|
||||||
}
|
}
|
||||||
if err := copyFromRoot(root, rel, dest); err != nil {
|
if err := copyFromRoot(root, rel, cfg.BrainDir, dest); err != nil {
|
||||||
return fmt.Errorf("copy to processed: %w", err)
|
return fmt.Errorf("copy to processed: %w", err)
|
||||||
}
|
}
|
||||||
if err := os.WriteFile(path+".processed", []byte(date), 0o644); err != nil {
|
if err := putBrainFile(cfg.BrainDir, "raw/"+filepath.ToSlash(rel)+".processed", []byte(date)); err != nil {
|
||||||
slog.Error("watcher: failed to write .processed marker", "error", err)
|
slog.Error("watcher: failed to write .processed marker", "error", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -214,23 +208,22 @@ func appendWatcherLog(brainDir, filename string, runErr error, date string) erro
|
|||||||
entry := fmt.Sprintf("## %s — watcher error\n\n- **File:** %s\n- **Error:** %s\n\n",
|
entry := fmt.Sprintf("## %s — watcher error\n\n- **File:** %s\n- **Error:** %s\n\n",
|
||||||
date, filename, runErr.Error())
|
date, filename, runErr.Error())
|
||||||
|
|
||||||
logPath := filepath.Join(brainDir, "log.md")
|
if err := brainfs.AppendFile(brainDir, "log.md", []byte(entry)); err != nil {
|
||||||
f, err := os.OpenFile(logPath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("open log: %w", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
if _, err = f.WriteString(entry); err != nil {
|
|
||||||
f.Close() //nolint:errcheck
|
|
||||||
return fmt.Errorf("write log: %w", err)
|
return fmt.Errorf("write log: %w", err)
|
||||||
}
|
}
|
||||||
return f.Close()
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// putBrainFile creates or replaces a brain-relative file through the brain
|
||||||
|
// root, never through a symlinked directory or file (hyperguild#158).
|
||||||
|
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
|
// 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
|
// that is not a regular file, so a swapped-in symlink cannot copy an outside
|
||||||
// file into raw/processed/ or raw/failed/ (hyperguild#153).
|
// file into raw/processed/ or raw/failed/ (hyperguild#153).
|
||||||
func copyFromRoot(root *os.Root, rel, dst string) error {
|
func copyFromRoot(root *os.Root, rel, brainDir, dstRel string) error {
|
||||||
fi, err := root.Lstat(rel)
|
fi, err := root.Lstat(rel)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
@@ -243,13 +236,9 @@ func copyFromRoot(root *os.Root, rel, dst string) error {
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
defer func() { _ = in.Close() }()
|
defer func() { _ = in.Close() }()
|
||||||
out, err := os.Create(dst)
|
data, err := io.ReadAll(in)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
if _, err := io.Copy(out, in); err != nil {
|
return putBrainFile(brainDir, dstRel, data)
|
||||||
_ = out.Close()
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
return out.Close()
|
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,66 @@
|
|||||||
|
package watcher
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
|
||||||
|
"github.com/mathiasbq/hyperguild/ingestion/internal/pipeline"
|
||||||
|
)
|
||||||
|
|
||||||
|
func okPipeline() pipeline.Config {
|
||||||
|
return pipeline.Config{Complete: func(context.Context, string, string) (string, error) {
|
||||||
|
b, _ := json.Marshal([]pipeline.RawPage{{Title: "X", Type: "source", Subtype: "article", Content: "## S\n\nx\n"}})
|
||||||
|
return string(b), nil
|
||||||
|
}, Schema: "# Schema"}
|
||||||
|
}
|
||||||
|
|
||||||
|
func failingPipeline() pipeline.Config {
|
||||||
|
return pipeline.Config{Complete: func(context.Context, string, string) (string, error) {
|
||||||
|
return "", errors.New("model down")
|
||||||
|
}, Schema: "# Schema"}
|
||||||
|
}
|
||||||
|
|
||||||
|
// hyperguild#158: the watcher wrote the archive copy under raw/processed or
|
||||||
|
// raw/failed, and its log.md entry, by path, so a symlinked archive
|
||||||
|
// directory or log redirected the write elsewhere in the brain or outside
|
||||||
|
// it. (A pre-planted marker symlink makes the watcher skip the note, so the
|
||||||
|
// marker write itself is reachable only by a race.)
|
||||||
|
func TestProcessDirDoesNotWriteThroughASymlinkedArchiveDirOrLog(t *testing.T) {
|
||||||
|
for name, pcfg := range map[string]pipeline.Config{"processed": okPipeline(), "failed": failingPipeline()} {
|
||||||
|
t.Run(name, func(t *testing.T) {
|
||||||
|
brainDir := setupBrainDir(t)
|
||||||
|
outside := t.TempDir()
|
||||||
|
victim := filepath.Join(brainDir, "wiki", "concepts", "v.md")
|
||||||
|
require.NoError(t, os.WriteFile(victim, []byte("untouched"), 0o644))
|
||||||
|
require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "n.md"), []byte("note"), 0o644))
|
||||||
|
require.NoError(t, os.Symlink(outside, filepath.Join(brainDir, "raw", name)))
|
||||||
|
if name == "failed" {
|
||||||
|
require.NoError(t, os.Symlink(filepath.Join("wiki", "concepts", "v.md"), filepath.Join(brainDir, "log.md")))
|
||||||
|
}
|
||||||
|
_ = processDir(context.Background(), Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: pcfg}, "2026-10-05")
|
||||||
|
got, _ := os.ReadFile(victim)
|
||||||
|
assert.Equal(t, "untouched", string(got), "the log was written through a symlink")
|
||||||
|
ents, _ := os.ReadDir(outside)
|
||||||
|
assert.Empty(t, ents, "the archive copy was written outside the brain")
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestProcessDirWritesMarkersAndArchiveInRealDirectories(t *testing.T) {
|
||||||
|
brainDir := setupBrainDir(t)
|
||||||
|
require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "n.md"), []byte("note"), 0o644))
|
||||||
|
errs := processDir(context.Background(), Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: okPipeline()}, "2026-10-05")
|
||||||
|
assert.Empty(t, errs)
|
||||||
|
_, err := os.Stat(filepath.Join(brainDir, "raw", "n.md.processed"))
|
||||||
|
assert.NoError(t, err, "positive control: marker written")
|
||||||
|
_, err = os.Stat(filepath.Join(brainDir, "raw", "processed", "2026-10-05", "n.md"))
|
||||||
|
assert.NoError(t, err, "positive control: archive copy written")
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user