Compare commits
3
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
31071bc9b1 | ||
|
|
66b6cf8a77 | ||
|
|
74a9621ba6 |
@@ -13,7 +13,9 @@ import (
|
||||
"io"
|
||||
"io/fs"
|
||||
"os"
|
||||
"path"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"syscall"
|
||||
)
|
||||
|
||||
@@ -53,6 +55,33 @@ func (b *Brain) Rel(path string) (string, error) {
|
||||
return filepath.ToSlash(rel), nil
|
||||
}
|
||||
|
||||
// Lstat stats a brain-relative path through the root without following a
|
||||
// final symlink.
|
||||
func (b *Brain) Lstat(rel string) (fs.FileInfo, error) {
|
||||
return b.root.Lstat(filepath.ToSlash(rel))
|
||||
}
|
||||
|
||||
// realDir returns ErrNotRegular unless every component of rel is a real
|
||||
// directory. A swap after the check is not caught (accepted residual,
|
||||
// hyperguild#157).
|
||||
func (b *Brain) realDir(rel string) error {
|
||||
if rel == "." {
|
||||
return nil
|
||||
}
|
||||
prefix := ""
|
||||
for _, part := range strings.Split(rel, "/") {
|
||||
prefix = path.Join(prefix, part)
|
||||
fi, err := b.root.Lstat(prefix)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !fi.IsDir() {
|
||||
return fmt.Errorf("%s: walk root component %s is not a directory: %w", rel, prefix, ErrNotRegular)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ReadRegular reads a brain-relative path, refusing anything that is not a
|
||||
// regular file, both before the open and on the opened handle.
|
||||
func (b *Brain) ReadRegular(rel string) ([]byte, error) {
|
||||
@@ -127,15 +156,34 @@ func (b *Brain) readOpenedMax(rel string, max int64) ([]byte, error) {
|
||||
}
|
||||
|
||||
// wrapAfterStat is a test seam: it runs between the size check and the read,
|
||||
// and wraps the reader the read consumes.
|
||||
// and wraps the reader the read consumes. A package global: brainfs tests
|
||||
// must not use t.Parallel.
|
||||
var wrapAfterStat func(io.Reader) io.Reader
|
||||
|
||||
// WalkRegular walks a brain-relative directory through the root and calls fn
|
||||
// for each regular file (directories are descended; symlinks, FIFOs and
|
||||
// devices are skipped, never followed). An entry error is passed to onErr and
|
||||
// the walk continues; a nil onErr ignores it.
|
||||
// for each regular file. Directories are descended; symlinks, FIFOs and
|
||||
// devices found by the walk are skipped. A walk root with any component that
|
||||
// is not a real directory goes to onErr as ErrNotRegular and nothing is
|
||||
// yielded. An entry error is passed to onErr and the walk continues; a nil
|
||||
// onErr ignores it.
|
||||
//
|
||||
// Accepted residual (hyperguild#157): a directory swapped for a symlink
|
||||
// between the listing and the read is followed. os.Root still keeps the read
|
||||
// inside the brain, but not inside relDir.
|
||||
func (b *Brain) WalkRegular(relDir string, fn func(rel string) error, onErr func(rel string, err error)) error {
|
||||
return fs.WalkDir(b.root.FS(), filepath.ToSlash(relDir), func(p string, d fs.DirEntry, err error) error {
|
||||
relDir = filepath.ToSlash(relDir)
|
||||
// fs.WalkDir stats its root with a follow, and os.Root allows a symlink
|
||||
// that stays inside the brain: knowledge -> . would alias the whole
|
||||
// brain. Require every component of the root to be a real directory
|
||||
// (Lstat checks only the last one).
|
||||
err := b.realDir(relDir)
|
||||
if err != nil {
|
||||
if onErr != nil {
|
||||
onErr(relDir, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
return fs.WalkDir(b.root.FS(), relDir, func(p string, d fs.DirEntry, err error) error {
|
||||
if err != nil {
|
||||
if onErr != nil {
|
||||
onErr(p, err)
|
||||
|
||||
@@ -3,6 +3,7 @@ package brainfs
|
||||
import (
|
||||
"errors"
|
||||
"io"
|
||||
"io/fs"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"syscall"
|
||||
@@ -139,3 +140,52 @@ func (c *countingReader) Read(p []byte) (int, error) {
|
||||
c.n += n
|
||||
return n, err
|
||||
}
|
||||
|
||||
func TestLstatDoesNotFollowSymlinksAndRefusesEscapes(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
outside := filepath.Join(t.TempDir(), "o.md")
|
||||
require.NoError(t, os.WriteFile(outside, []byte("o"), 0o644))
|
||||
require.NoError(t, os.Symlink(outside, filepath.Join(dir, "l.md")))
|
||||
b, err := Open(dir)
|
||||
require.NoError(t, err)
|
||||
defer func() { _ = b.Close() }()
|
||||
fi, err := b.Lstat("l.md")
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, fs.ModeSymlink, fi.Mode().Type())
|
||||
_, err = b.Lstat("../" + filepath.Base(filepath.Dir(outside)) + "/o.md")
|
||||
assert.Error(t, err)
|
||||
}
|
||||
|
||||
func TestWalkRegularReportsASymlinkedWalkRootAndYieldsNothing(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
require.NoError(t, os.MkdirAll(filepath.Join(dir, "real"), 0o755))
|
||||
require.NoError(t, os.WriteFile(filepath.Join(dir, "real", "n.md"), []byte("n"), 0o644))
|
||||
require.NoError(t, os.Symlink("real", filepath.Join(dir, "alias")))
|
||||
b, err := Open(dir)
|
||||
require.NoError(t, err)
|
||||
defer func() { _ = b.Close() }()
|
||||
var got []string
|
||||
var errs []error
|
||||
require.NoError(t, b.WalkRegular("alias", func(rel string) error { got = append(got, rel); return nil },
|
||||
func(_ string, err error) { errs = append(errs, err) }))
|
||||
assert.Empty(t, got)
|
||||
require.Len(t, errs, 1)
|
||||
assert.ErrorIs(t, errs[0], ErrNotRegular)
|
||||
}
|
||||
|
||||
func TestWalkRegularReportsASymlinkedIntermediateComponentOfTheWalkRoot(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
require.NoError(t, os.MkdirAll(filepath.Join(dir, "real", "sub"), 0o755))
|
||||
require.NoError(t, os.WriteFile(filepath.Join(dir, "real", "sub", "n.md"), []byte("n"), 0o644))
|
||||
require.NoError(t, os.Symlink("real", filepath.Join(dir, "alias")))
|
||||
b, err := Open(dir)
|
||||
require.NoError(t, err)
|
||||
defer func() { _ = b.Close() }()
|
||||
var got []string
|
||||
var errs []error
|
||||
require.NoError(t, b.WalkRegular("alias/sub", func(rel string) error { got = append(got, rel); return nil },
|
||||
func(_ string, err error) { errs = append(errs, err) }))
|
||||
assert.Empty(t, got)
|
||||
require.Len(t, errs, 1)
|
||||
assert.ErrorIs(t, errs[0], ErrNotRegular)
|
||||
}
|
||||
|
||||
@@ -61,6 +61,31 @@ func TestQueryDoesNotFollowASymlinkedDirectory(t *testing.T) {
|
||||
assert.Zero(t, leaked(rs))
|
||||
}
|
||||
|
||||
// A search root that is a symlink staying inside the brain (wiki/<w> -> ..)
|
||||
// was followed by fs.WalkDir, aliasing raw/ and root files into results
|
||||
// (PR #166 review). The root is now Lstat-checked and skipped.
|
||||
func TestQueryDoesNotFollowASymlinkedWingRootInsideTheBrain(t *testing.T) {
|
||||
brain, _, _ := searchBrain(t)
|
||||
require.NoError(t, os.MkdirAll(filepath.Join(brain, "raw"), 0o755))
|
||||
require.NoError(t, os.WriteFile(filepath.Join(brain, "raw", "r.md"), []byte("# r\n\nzqrawonly"), 0o644))
|
||||
require.NoError(t, os.Symlink("..", filepath.Join(brain, "wiki", "alias")))
|
||||
rs, err := Query(brain, QueryOptions{Query: "zqrawonly", Wing: "alias", Limit: 10})
|
||||
require.NoError(t, err)
|
||||
assert.Empty(t, rs)
|
||||
}
|
||||
|
||||
// With a hall the search root is wiki/<wing>/<hall>; a symlinked <wing> with
|
||||
// a real <hall> directory under its target was followed (PR #166 round 2).
|
||||
func TestQueryDoesNotFollowASymlinkedWingUnderAWingAndHallRoot(t *testing.T) {
|
||||
brain, _, _ := searchBrain(t)
|
||||
require.NoError(t, os.MkdirAll(filepath.Join(brain, "raw", "facts"), 0o755))
|
||||
require.NoError(t, os.WriteFile(filepath.Join(brain, "raw", "facts", "r.md"), []byte("# r\n\nzqrawhall"), 0o644))
|
||||
require.NoError(t, os.Symlink(filepath.Join("..", "raw"), filepath.Join(brain, "wiki", "alias")))
|
||||
rs, err := Query(brain, QueryOptions{Query: "zqrawhall", Wing: "alias", Hall: "facts", Limit: 10})
|
||||
require.NoError(t, err)
|
||||
assert.Empty(t, rs)
|
||||
}
|
||||
|
||||
// swapInto keeps renaming a regular note, and either a symlink to outside or
|
||||
// a FIFO, into path. Each FIFO has a writer held open (as any local process
|
||||
// can): a FIFO with no writer reads EOF at once, which would let a reader
|
||||
|
||||
@@ -2,12 +2,15 @@ package vectorstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io/fs"
|
||||
"log/slog"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"path"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
|
||||
)
|
||||
|
||||
// Embedder produces dense vectors. The embed package's Client satisfies
|
||||
@@ -90,31 +93,38 @@ func Sync(ctx context.Context, brainDir string, store Store, embedder Embedder)
|
||||
}
|
||||
seenParents := make(map[string]struct{})
|
||||
|
||||
// Through the brain root, regular files only (hyperguild#157): the brain
|
||||
// is a Syncthing folder, and os.ReadFile followed a synced symlink into
|
||||
// the embeddings and blocked on a FIFO.
|
||||
bfs, err := brainfs.Open(brainDir)
|
||||
if err != nil {
|
||||
return res, fmt.Errorf("open brain: %w", err)
|
||||
}
|
||||
defer func() { _ = bfs.Close() }()
|
||||
|
||||
for _, sub := range scanDirs {
|
||||
root := filepath.Join(brainDir, sub)
|
||||
if _, err := os.Stat(root); os.IsNotExist(err) {
|
||||
if _, err := bfs.Lstat(sub); errors.Is(err, fs.ErrNotExist) {
|
||||
continue
|
||||
}
|
||||
|
||||
err = filepath.WalkDir(root, func(path string, d os.DirEntry, err error) error {
|
||||
if err != nil {
|
||||
return err
|
||||
// Any other walk error, including a scan root that is a symlink,
|
||||
// aborts the run below.
|
||||
var walkErr error
|
||||
onErr := func(rel string, err error) {
|
||||
if walkErr == nil {
|
||||
walkErr = fmt.Errorf("%s: %w", rel, err)
|
||||
}
|
||||
if d.IsDir() || !strings.HasSuffix(path, ".md") || d.Name() == "_index.md" {
|
||||
}
|
||||
err = bfs.WalkRegular(sub, func(relSlash string) error {
|
||||
if !strings.HasSuffix(relSlash, ".md") || path.Base(relSlash) == "_index.md" {
|
||||
return nil
|
||||
}
|
||||
rel, err := filepath.Rel(brainDir, path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
relSlash := filepath.ToSlash(rel)
|
||||
seenParents[relSlash] = struct{}{}
|
||||
|
||||
if ps, ok := parents[relSlash]; ok {
|
||||
// File already has chunks in the store. Re-embed only when
|
||||
// the file has been edited since the oldest chunk was
|
||||
// written. Tolerate clock skew with a sub-second grace.
|
||||
info, statErr := d.Info()
|
||||
info, statErr := bfs.Lstat(relSlash)
|
||||
if statErr != nil {
|
||||
res.Errors = append(res.Errors, fmt.Errorf("stat %s: %w", relSlash, statErr))
|
||||
return nil
|
||||
@@ -132,7 +142,7 @@ func Sync(ctx context.Context, brainDir string, store Store, embedder Embedder)
|
||||
}
|
||||
}
|
||||
|
||||
content, readErr := os.ReadFile(path)
|
||||
content, readErr := bfs.ReadRegular(relSlash)
|
||||
if readErr != nil {
|
||||
res.Errors = append(res.Errors, fmt.Errorf("read %s: %w", relSlash, readErr))
|
||||
return nil
|
||||
@@ -151,7 +161,12 @@ func Sync(ctx context.Context, brainDir string, store Store, embedder Embedder)
|
||||
res.Added++
|
||||
}
|
||||
return nil
|
||||
})
|
||||
}, onErr)
|
||||
if err == nil {
|
||||
err = walkErr
|
||||
}
|
||||
// Abort before the delete pass: an incomplete walk would leave
|
||||
// seenParents short and delete chunks of files that still exist.
|
||||
if err != nil {
|
||||
return res, fmt.Errorf("walk %s: %w", sub, err)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,126 @@
|
||||
package vectorstore_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs/brainfstest"
|
||||
"github.com/mathiasbq/hyperguild/ingestion/internal/vectorstore"
|
||||
)
|
||||
|
||||
// recordingEmbedder keeps every text it was asked to embed.
|
||||
type recordingEmbedder struct {
|
||||
mu sync.Mutex
|
||||
texts []string
|
||||
}
|
||||
|
||||
func (e *recordingEmbedder) Embed(_ context.Context, text string) ([]float32, error) {
|
||||
e.mu.Lock()
|
||||
defer e.mu.Unlock()
|
||||
e.texts = append(e.texts, text)
|
||||
return []float32{1}, nil
|
||||
}
|
||||
|
||||
func (e *recordingEmbedder) all() string {
|
||||
e.mu.Lock()
|
||||
defer e.mu.Unlock()
|
||||
return strings.Join(e.texts, "\n")
|
||||
}
|
||||
|
||||
// hyperguild#157 slice 3: Sync walked wiki/ and knowledge/ and read each .md
|
||||
// with os.ReadFile, so a symlinked note had its target embedded and served by
|
||||
// semantic search, and a FIFO stalled the sync.
|
||||
func TestSyncRefusesSymlinkedNotesAndSurvivesSwapRaces(t *testing.T) {
|
||||
brain := t.TempDir()
|
||||
writeNote(t, brain, "wiki/w/facts/ok.md", "inside note")
|
||||
require.NoError(t, os.Symlink(brainfstest.Outside(t), filepath.Join(brain, "wiki", "w", "facts", "leak.md")))
|
||||
|
||||
emb := &recordingEmbedder{}
|
||||
store := &stubStore{}
|
||||
_, err := vectorstore.Sync(context.Background(), brain, store, emb)
|
||||
require.NoError(t, err)
|
||||
assert.Contains(t, emb.all(), "inside note", "positive control: the regular note is embedded")
|
||||
assert.NotContains(t, emb.all(), brainfstest.Sentinel)
|
||||
for p := range store.upserts {
|
||||
assert.NotContains(t, p, "leak.md")
|
||||
}
|
||||
|
||||
p := filepath.Join(brain, "knowledge", "r.md")
|
||||
require.NoError(t, os.MkdirAll(filepath.Dir(p), 0o755))
|
||||
sync := func() string {
|
||||
e := &recordingEmbedder{}
|
||||
_, _ = vectorstore.Sync(context.Background(), brain, &stubStore{}, e)
|
||||
return e.all()
|
||||
}
|
||||
brainfstest.AssertNoLeak(t, p, "inside", 1500*time.Millisecond, sync)
|
||||
brainfstest.AssertNoHang(t, p, "inside", 1500*time.Millisecond, 6*time.Second, func() { _ = sync() })
|
||||
}
|
||||
|
||||
// Chunks embedded from a symlinked note before this fix are deleted: the
|
||||
// note is no longer seen, so the delete pass drops its rows.
|
||||
func TestSyncDeletesChunksOfASymlinkedNote(t *testing.T) {
|
||||
brain := t.TempDir()
|
||||
writeNote(t, brain, "wiki/w/facts/ok.md", "inside note")
|
||||
require.NoError(t, os.Symlink(brainfstest.Outside(t), filepath.Join(brain, "wiki", "w", "facts", "leak.md")))
|
||||
store := &stubStore{known: map[string]time.Time{
|
||||
"wiki/w/facts/leak.md#0001": {},
|
||||
"wiki/w/facts/ok.md#0001": {},
|
||||
}}
|
||||
_, err := vectorstore.Sync(context.Background(), brain, store, &recordingEmbedder{})
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, []string{"wiki/w/facts/leak.md#0001"}, store.deletes)
|
||||
}
|
||||
|
||||
// A walk error aborts Sync before the delete pass, so chunks of files the
|
||||
// walk could not see are kept.
|
||||
func TestSyncAbortsBeforeDeletingWhenASubdirIsUnreadable(t *testing.T) {
|
||||
if os.Geteuid() == 0 {
|
||||
t.Skip("root ignores directory permissions")
|
||||
}
|
||||
brain := t.TempDir()
|
||||
writeNote(t, brain, "wiki/w/facts/ok.md", "inside note")
|
||||
locked := filepath.Join(brain, "wiki", "w")
|
||||
require.NoError(t, os.Chmod(locked, 0o000))
|
||||
t.Cleanup(func() { _ = os.Chmod(locked, 0o755) })
|
||||
store := &stubStore{known: map[string]time.Time{"wiki/w/facts/ok.md#0001": {}}}
|
||||
_, err := vectorstore.Sync(context.Background(), brain, store, &recordingEmbedder{})
|
||||
assert.Error(t, err)
|
||||
assert.Empty(t, store.deletes)
|
||||
}
|
||||
|
||||
// A scan root that is a symlink inside the brain (knowledge -> .) was
|
||||
// followed by fs.WalkDir, aliasing raw/, sessions/ and root files into the
|
||||
// index. Sync now aborts on it, before embedding or deleting anything.
|
||||
func TestSyncAbortsOnASymlinkedScanRoot(t *testing.T) {
|
||||
brain := t.TempDir()
|
||||
writeNote(t, brain, "wiki/w/a.md", "wiki note")
|
||||
writeNote(t, brain, "sessions/s1.md", "session note")
|
||||
require.NoError(t, os.Symlink(".", filepath.Join(brain, "knowledge")))
|
||||
emb := &recordingEmbedder{}
|
||||
store := &stubStore{known: map[string]time.Time{"wiki/w/a.md#0001": {}}}
|
||||
_, err := vectorstore.Sync(context.Background(), brain, store, emb)
|
||||
assert.Error(t, err)
|
||||
assert.NotContains(t, emb.all(), "session note")
|
||||
for p := range store.upserts {
|
||||
assert.False(t, strings.HasPrefix(p, "knowledge/"), "upserted %s", p)
|
||||
}
|
||||
assert.Empty(t, store.deletes)
|
||||
}
|
||||
|
||||
// A missing scan root is skipped; the other root still syncs.
|
||||
func TestSyncSkipsAMissingScanRoot(t *testing.T) {
|
||||
brain := t.TempDir()
|
||||
writeNote(t, brain, "wiki/w/a.md", "wiki note")
|
||||
emb := &recordingEmbedder{}
|
||||
_, err := vectorstore.Sync(context.Background(), brain, &stubStore{}, emb)
|
||||
require.NoError(t, err)
|
||||
assert.Contains(t, emb.all(), "wiki note")
|
||||
}
|
||||
Reference in New Issue
Block a user