Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
4c7fcb6bdc | ||
|
|
4a1abafd8f |
@@ -3,18 +3,15 @@ package api
|
|||||||
import (
|
import (
|
||||||
"crypto/sha256"
|
"crypto/sha256"
|
||||||
"encoding/hex"
|
"encoding/hex"
|
||||||
"errors"
|
|
||||||
"fmt"
|
"fmt"
|
||||||
"io/fs"
|
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/brain"
|
"github.com/mathiasbq/hyperguild/ingestion/internal/brain"
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
|
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/claim"
|
"github.com/mathiasbq/hyperguild/ingestion/internal/claim"
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/notes"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
// ContentHash returns the lowercase hex sha256 of b. It is the note's
|
// ContentHash returns the lowercase hex sha256 of b. It is the note's
|
||||||
@@ -94,23 +91,14 @@ func UpdateNote(brainDir string, opts UpdateNoteOptions) (relPath, contentHash,
|
|||||||
rel = filepath.ToSlash(rel)
|
rel = filepath.ToSlash(rel)
|
||||||
}
|
}
|
||||||
|
|
||||||
// The path is caller-supplied (brain_update): resolve it as brain_edit
|
rel, abs, err := resolveWithin(brainDir, rel)
|
||||||
// does, knowledge/ and wiki/ notes only, and read and rewrite it through
|
|
||||||
// the brain root, never through a symlink (hyperguild#169).
|
|
||||||
t, err := notes.Resolve(brainDir, rel, notes.ScopeNotes)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", "", "", err
|
return "", "", "", err
|
||||||
}
|
}
|
||||||
rel = t.Rel
|
|
||||||
bfs, err := brainfs.Open(brainDir)
|
|
||||||
if err != nil {
|
|
||||||
return "", "", "", fmt.Errorf("open brain: %w", err)
|
|
||||||
}
|
|
||||||
defer func() { _ = bfs.Close() }()
|
|
||||||
|
|
||||||
prior, err := bfs.ReadRegular(rel)
|
prior, err := os.ReadFile(abs)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if errors.Is(err, fs.ErrNotExist) {
|
if os.IsNotExist(err) {
|
||||||
return "", "", "", fmt.Errorf("note %q does not exist: use brain_write to create", rel)
|
return "", "", "", fmt.Errorf("note %q does not exist: use brain_write to create", rel)
|
||||||
}
|
}
|
||||||
return "", "", "", fmt.Errorf("read target: %w", err)
|
return "", "", "", fmt.Errorf("read target: %w", err)
|
||||||
@@ -176,7 +164,7 @@ func UpdateNote(brainDir string, opts UpdateNoteOptions) (relPath, contentHash,
|
|||||||
return "", "", "", err
|
return "", "", "", err
|
||||||
}
|
}
|
||||||
out := []byte(stamped)
|
out := []byte(stamped)
|
||||||
if err := bfs.WriteRegular(rel, out); err != nil {
|
if err := os.WriteFile(abs, out, 0o644); err != nil {
|
||||||
return "", "", "", fmt.Errorf("write: %w", err)
|
return "", "", "", fmt.Errorf("write: %w", err)
|
||||||
}
|
}
|
||||||
return rel, ContentHash(out), priorHash, nil
|
return rel, ContentHash(out), priorHash, nil
|
||||||
|
|||||||
@@ -1,112 +0,0 @@
|
|||||||
package api_test
|
|
||||||
|
|
||||||
import (
|
|
||||||
"os"
|
|
||||||
"path/filepath"
|
|
||||||
"sync"
|
|
||||||
"sync/atomic"
|
|
||||||
"testing"
|
|
||||||
"time"
|
|
||||||
|
|
||||||
"github.com/stretchr/testify/assert"
|
|
||||||
"github.com/stretchr/testify/require"
|
|
||||||
|
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/api"
|
|
||||||
)
|
|
||||||
|
|
||||||
// brain_update passed the caller's path through resolveWithin, a lexical
|
|
||||||
// containment check only: any file in the brain could be superseded
|
|
||||||
// (classification.yaml, raw/, sessions/), and a final symlink was written
|
|
||||||
// through. It now resolves like brain_edit: knowledge/ and wiki/ notes only,
|
|
||||||
// read and rewritten through the brain root.
|
|
||||||
func TestUpdateNoteRefusesPathsOutsideKnowledgeAndWikiNotes(t *testing.T) {
|
|
||||||
brain := t.TempDir()
|
|
||||||
files := map[string]string{
|
|
||||||
"classification.yaml": "levels: {}\n",
|
|
||||||
"raw/r.md": "raw\n",
|
|
||||||
"sessions/s.jsonl": "{}\n",
|
|
||||||
"wiki/w/facts/ok.md": "---\ntitle: ok\n---\nold\n",
|
|
||||||
"knowledge/k.md": "---\ntitle: k\n---\nold\n",
|
|
||||||
"wiki/w/facts/x.txt": "txt\n",
|
|
||||||
}
|
|
||||||
for rel, body := range files {
|
|
||||||
p := filepath.Join(brain, filepath.FromSlash(rel))
|
|
||||||
require.NoError(t, os.MkdirAll(filepath.Dir(p), 0o755))
|
|
||||||
require.NoError(t, os.WriteFile(p, []byte(body), 0o644))
|
|
||||||
}
|
|
||||||
for _, rel := range []string{"classification.yaml", "raw/r.md", "sessions/s.jsonl", "wiki/w/facts/x.txt", "wiki/w/../../raw/r.md"} {
|
|
||||||
_, _, _, err := api.UpdateNote(brain, api.UpdateNoteOptions{Path: rel, Content: "replaced"})
|
|
||||||
assert.Error(t, err, rel)
|
|
||||||
}
|
|
||||||
for rel, body := range files {
|
|
||||||
if rel == "wiki/w/facts/ok.md" || rel == "knowledge/k.md" {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
got, err := os.ReadFile(filepath.Join(brain, filepath.FromSlash(rel)))
|
|
||||||
require.NoError(t, err)
|
|
||||||
assert.Equal(t, body, string(got), rel)
|
|
||||||
}
|
|
||||||
for _, rel := range []string{"wiki/w/facts/ok.md", "knowledge/k.md"} {
|
|
||||||
_, _, _, err := api.UpdateNote(brain, api.UpdateNoteOptions{Path: rel, Content: "new body"})
|
|
||||||
require.NoError(t, err, "positive control: %s", rel)
|
|
||||||
got, _ := os.ReadFile(filepath.Join(brain, filepath.FromSlash(rel)))
|
|
||||||
assert.Contains(t, string(got), "new body")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestUpdateNoteDoesNotWriteThroughASymlinkedNote(t *testing.T) {
|
|
||||||
brain := t.TempDir()
|
|
||||||
facts := filepath.Join(brain, "wiki", "w", "facts")
|
|
||||||
require.NoError(t, os.MkdirAll(facts, 0o755))
|
|
||||||
victim := filepath.Join(brain, "raw", "victim.md")
|
|
||||||
require.NoError(t, os.MkdirAll(filepath.Dir(victim), 0o755))
|
|
||||||
require.NoError(t, os.WriteFile(victim, []byte("untouched\n"), 0o644))
|
|
||||||
require.NoError(t, os.Symlink(filepath.Join("..", "..", "..", "raw", "victim.md"), filepath.Join(facts, "l.md")))
|
|
||||||
_, _, _, err := api.UpdateNote(brain, api.UpdateNoteOptions{Path: "wiki/w/facts/l.md", Content: "replaced"})
|
|
||||||
assert.Error(t, err)
|
|
||||||
got, _ := os.ReadFile(victim)
|
|
||||||
assert.Equal(t, "untouched\n", string(got))
|
|
||||||
}
|
|
||||||
|
|
||||||
// The note is swapped between a regular file and an in-brain symlink to a
|
|
||||||
// raw/ file while UpdateNote runs: the read and the write each refuse a
|
|
||||||
// symlink, so the raw/ file is never rewritten.
|
|
||||||
func TestUpdateNoteNeverRewritesASymlinkSwappedInForTheNote(t *testing.T) {
|
|
||||||
brain := t.TempDir()
|
|
||||||
facts := filepath.Join(brain, "wiki", "w", "facts")
|
|
||||||
require.NoError(t, os.MkdirAll(facts, 0o755))
|
|
||||||
victim := filepath.Join(brain, "raw", "victim.md")
|
|
||||||
require.NoError(t, os.MkdirAll(filepath.Dir(victim), 0o755))
|
|
||||||
require.NoError(t, os.WriteFile(victim, []byte("untouched\n"), 0o644))
|
|
||||||
note := filepath.Join(facts, "n.md")
|
|
||||||
body := "---\ntitle: n\n---\nold\n"
|
|
||||||
require.NoError(t, os.WriteFile(note, []byte(body), 0o644))
|
|
||||||
|
|
||||||
var stop atomic.Bool
|
|
||||||
var wg sync.WaitGroup
|
|
||||||
wg.Add(1)
|
|
||||||
go func() {
|
|
||||||
defer wg.Done()
|
|
||||||
tmp := note + ".swap"
|
|
||||||
for i := 0; !stop.Load(); i++ {
|
|
||||||
_ = os.Remove(tmp)
|
|
||||||
if i%2 == 0 {
|
|
||||||
_ = os.WriteFile(tmp, []byte(body), 0o644)
|
|
||||||
} else {
|
|
||||||
_ = os.Symlink(filepath.Join("..", "..", "..", "raw", "victim.md"), tmp)
|
|
||||||
}
|
|
||||||
_ = os.Rename(tmp, note)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
ok := 0
|
|
||||||
for deadline := time.Now().Add(1500 * time.Millisecond); time.Now().Before(deadline); {
|
|
||||||
if _, _, _, err := api.UpdateNote(brain, api.UpdateNoteOptions{Path: "wiki/w/facts/n.md", Content: "new"}); err == nil {
|
|
||||||
ok++
|
|
||||||
}
|
|
||||||
}
|
|
||||||
stop.Store(true)
|
|
||||||
wg.Wait()
|
|
||||||
got, _ := os.ReadFile(victim)
|
|
||||||
assert.Equal(t, "untouched\n", string(got))
|
|
||||||
assert.Positive(t, ok, "positive control: some updates reach the regular note")
|
|
||||||
}
|
|
||||||
@@ -4,12 +4,9 @@ import (
|
|||||||
"bufio"
|
"bufio"
|
||||||
"fmt"
|
"fmt"
|
||||||
"os"
|
"os"
|
||||||
"path"
|
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
// seeAlsoHeader is the markdown heading used to group cross-wing links.
|
// seeAlsoHeader is the markdown heading used to group cross-wing links.
|
||||||
@@ -200,12 +197,6 @@ func fileExists(p string) bool {
|
|||||||
// section headers. Rejects same-wing pairs (a tunnel is by definition
|
// section headers. Rejects same-wing pairs (a tunnel is by definition
|
||||||
// cross-wing) and missing notes.
|
// cross-wing) and missing notes.
|
||||||
func WriteTunnel(brainDir, sourcePath, targetPath string) error {
|
func WriteTunnel(brainDir, sourcePath, targetPath string) error {
|
||||||
if err := checkNotePath(sourcePath); err != nil {
|
|
||||||
return fmt.Errorf("source: %w", err)
|
|
||||||
}
|
|
||||||
if err := checkNotePath(targetPath); err != nil {
|
|
||||||
return fmt.Errorf("target: %w", err)
|
|
||||||
}
|
|
||||||
srcWing, err := wingOf(sourcePath)
|
srcWing, err := wingOf(sourcePath)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("source: %w", err)
|
return fmt.Errorf("source: %w", err)
|
||||||
@@ -218,50 +209,20 @@ func WriteTunnel(brainDir, sourcePath, targetPath string) error {
|
|||||||
return fmt.Errorf("tunnel must cross wings; got both in %q", srcWing)
|
return fmt.Errorf("tunnel must cross wings; got both in %q", srcWing)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Both notes are read and rewritten through the brain root, regular
|
srcFull := filepath.Join(brainDir, filepath.FromSlash(sourcePath))
|
||||||
// files only: brain_tunnel passes caller-supplied paths, and the brain is
|
tgtFull := filepath.Join(brainDir, filepath.FromSlash(targetPath))
|
||||||
// a Syncthing folder that can hold symlinks.
|
if _, err := os.Stat(srcFull); err != nil {
|
||||||
bfs, err := brainfs.Open(brainDir)
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("open brain: %w", err)
|
|
||||||
}
|
|
||||||
defer func() { _ = bfs.Close() }()
|
|
||||||
srcContent, err := bfs.ReadRegular(sourcePath)
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("source note: %w", err)
|
return fmt.Errorf("source note: %w", err)
|
||||||
}
|
}
|
||||||
tgtContent, err := bfs.ReadRegular(targetPath)
|
if _, err := os.Stat(tgtFull); err != nil {
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("target note: %w", err)
|
return fmt.Errorf("target note: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
if out, changed, err := withSeeAlso(srcContent, wikilinkOf(targetPath)); err != nil {
|
if err := appendSeeAlso(srcFull, wikilinkOf(targetPath)); err != nil {
|
||||||
return fmt.Errorf("update source: %w", err)
|
|
||||||
} else if changed {
|
|
||||||
if err := bfs.WriteRegular(sourcePath, out); err != nil {
|
|
||||||
return fmt.Errorf("update source: %w", err)
|
return fmt.Errorf("update source: %w", err)
|
||||||
}
|
}
|
||||||
}
|
if err := appendSeeAlso(tgtFull, wikilinkOf(sourcePath)); err != nil {
|
||||||
if out, changed, err := withSeeAlso(tgtContent, wikilinkOf(sourcePath)); err != nil {
|
|
||||||
return fmt.Errorf("update target: %w", err)
|
return fmt.Errorf("update target: %w", err)
|
||||||
} else if changed {
|
|
||||||
if err := bfs.WriteRegular(targetPath, out); err != nil {
|
|
||||||
return fmt.Errorf("update target: %w", err)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// checkNotePath accepts only a clean, local wiki/<wing>/<hall>/.../<slug>.md
|
|
||||||
// path: no "." or ".." segments, no empty segments, no leading slash.
|
|
||||||
func checkNotePath(p string) error {
|
|
||||||
if p == "" || path.Clean(p) != p || !filepath.IsLocal(p) || !strings.HasSuffix(p, ".md") {
|
|
||||||
return fmt.Errorf("not a clean wiki note path: %q", p)
|
|
||||||
}
|
|
||||||
for _, part := range strings.Split(p, "/") {
|
|
||||||
if part == "" || part == "." || part == ".." {
|
|
||||||
return fmt.Errorf("not a clean wiki note path: %q", p)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
@@ -287,13 +248,16 @@ func wingOf(relPath string) (string, error) {
|
|||||||
return parts[1], nil
|
return parts[1], nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// withSeeAlso returns content with `- [[link]]` under its See also section,
|
// appendSeeAlso inserts `- [[link]]` under the file's See also section,
|
||||||
// creating the section if absent. changed is false when the link is already
|
// creating the section if absent. No-op when the link is already present.
|
||||||
// present.
|
func appendSeeAlso(filePath, link string) error {
|
||||||
func withSeeAlso(content []byte, link string) (out []byte, changed bool, err error) {
|
content, err := os.ReadFile(filePath)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
wikilink := "[[" + link + "]]"
|
wikilink := "[[" + link + "]]"
|
||||||
if strings.Contains(string(content), wikilink) {
|
if strings.Contains(string(content), wikilink) {
|
||||||
return content, false, nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
bullet := "- " + wikilink
|
bullet := "- " + wikilink
|
||||||
@@ -302,7 +266,8 @@ func withSeeAlso(content []byte, link string) (out []byte, changed bool, err err
|
|||||||
// No section yet — append a fresh one. Always emit a trailing
|
// No section yet — append a fresh one. Always emit a trailing
|
||||||
// newline so subsequent appends don't merge into the previous line.
|
// newline so subsequent appends don't merge into the previous line.
|
||||||
trimmed := strings.TrimRight(string(content), "\n")
|
trimmed := strings.TrimRight(string(content), "\n")
|
||||||
return []byte(trimmed + "\n\n" + seeAlsoHeader + "\n\n" + bullet + "\n"), true, nil
|
out := trimmed + "\n\n" + seeAlsoHeader + "\n\n" + bullet + "\n"
|
||||||
|
return os.WriteFile(filePath, []byte(out), 0o644)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Section exists — splice the bullet in just before the next `## `
|
// Section exists — splice the bullet in just before the next `## `
|
||||||
@@ -328,11 +293,12 @@ func withSeeAlso(content []byte, link string) (out []byte, changed bool, err err
|
|||||||
b.WriteByte('\n')
|
b.WriteByte('\n')
|
||||||
}
|
}
|
||||||
if err := scanner.Err(); err != nil {
|
if err := scanner.Err(); err != nil {
|
||||||
return nil, false, err
|
return err
|
||||||
}
|
}
|
||||||
if !inserted {
|
if !inserted {
|
||||||
// section was the last thing in the file — just append bullet
|
// section was the last thing in the file — just append bullet
|
||||||
return []byte(strings.TrimRight(b.String(), "\n") + "\n" + bullet + "\n"), true, nil
|
out := strings.TrimRight(b.String(), "\n") + "\n" + bullet + "\n"
|
||||||
|
return os.WriteFile(filePath, []byte(out), 0o644)
|
||||||
}
|
}
|
||||||
return []byte(b.String()), true, nil
|
return os.WriteFile(filePath, []byte(b.String()), 0o644)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,141 +0,0 @@
|
|||||||
package brain_test
|
|
||||||
|
|
||||||
import (
|
|
||||||
"os"
|
|
||||||
"path/filepath"
|
|
||||||
"strings"
|
|
||||||
"sync"
|
|
||||||
"sync/atomic"
|
|
||||||
"testing"
|
|
||||||
"time"
|
|
||||||
|
|
||||||
"github.com/stretchr/testify/assert"
|
|
||||||
"github.com/stretchr/testify/require"
|
|
||||||
|
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/brain"
|
|
||||||
)
|
|
||||||
|
|
||||||
// brain_tunnel passes caller-supplied source and target paths to WriteTunnel,
|
|
||||||
// which checked only "wiki/<wing>/..." and then joined them onto the brain
|
|
||||||
// dir: ".." segments resolved outside the brain and appendSeeAlso rewrote the
|
|
||||||
// file there.
|
|
||||||
func TestWriteTunnelRefusesATargetThatTraversesOutOfTheBrain(t *testing.T) {
|
|
||||||
brain0 := t.TempDir()
|
|
||||||
src := filepath.Join(brain0, "wiki", "a", "facts", "s.md")
|
|
||||||
require.NoError(t, os.MkdirAll(filepath.Dir(src), 0o755))
|
|
||||||
require.NoError(t, os.WriteFile(src, []byte("---\ntitle: S\n---\nbody\n"), 0o644))
|
|
||||||
victim := filepath.Join(t.TempDir(), "victim.md")
|
|
||||||
require.NoError(t, os.WriteFile(victim, []byte("untouched\n"), 0o644))
|
|
||||||
|
|
||||||
up := strings.Repeat("../", strings.Count(filepath.Clean(brain0), "/")+3)
|
|
||||||
target := "wiki/b/x/" + up + strings.TrimPrefix(victim, "/")
|
|
||||||
err := brain.WriteTunnel(brain0, "wiki/a/facts/s.md", target)
|
|
||||||
assert.Error(t, err)
|
|
||||||
got, rerr := os.ReadFile(victim)
|
|
||||||
require.NoError(t, rerr)
|
|
||||||
assert.Equal(t, "untouched\n", string(got), "a file outside the brain was modified")
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestWriteTunnelRefusesUncleanOrNonWikiPaths(t *testing.T) {
|
|
||||||
brain0 := t.TempDir()
|
|
||||||
for _, p := range []string{"wiki/a/facts/s.md", "wiki/b/facts/t.md", "wiki/b/facts/t.txt", "raw/r.md"} {
|
|
||||||
require.NoError(t, os.MkdirAll(filepath.Dir(filepath.Join(brain0, p)), 0o755))
|
|
||||||
require.NoError(t, os.WriteFile(filepath.Join(brain0, p), []byte("x\n"), 0o644))
|
|
||||||
}
|
|
||||||
for _, target := range []string{
|
|
||||||
"wiki/b/facts/../../../raw/r.md",
|
|
||||||
"wiki/b/./facts/t.md",
|
|
||||||
"wiki/b//facts/t.md",
|
|
||||||
"/wiki/b/facts/t.md",
|
|
||||||
"wiki/b/facts/t.txt",
|
|
||||||
"wiki/b/facts/",
|
|
||||||
} {
|
|
||||||
assert.Error(t, brain.WriteTunnel(brain0, "wiki/a/facts/s.md", target), target)
|
|
||||||
}
|
|
||||||
for _, p := range []string{"raw/r.md", "wiki/b/facts/t.txt"} {
|
|
||||||
got, err := os.ReadFile(filepath.Join(brain0, p))
|
|
||||||
require.NoError(t, err)
|
|
||||||
assert.Equal(t, "x\n", string(got), p)
|
|
||||||
}
|
|
||||||
require.NoError(t, brain.WriteTunnel(brain0, "wiki/a/facts/s.md", "wiki/b/facts/t.md"), "positive control")
|
|
||||||
got, err := os.ReadFile(filepath.Join(brain0, "wiki", "b", "facts", "t.md"))
|
|
||||||
require.NoError(t, err)
|
|
||||||
assert.Contains(t, string(got), "[[a/facts/s]]")
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestWriteTunnelDoesNotWriteThroughASymlinkedNote(t *testing.T) {
|
|
||||||
brain0 := t.TempDir()
|
|
||||||
src := filepath.Join(brain0, "wiki", "a", "facts", "s.md")
|
|
||||||
require.NoError(t, os.MkdirAll(filepath.Dir(src), 0o755))
|
|
||||||
require.NoError(t, os.WriteFile(src, []byte("x\n"), 0o644))
|
|
||||||
tdir := filepath.Join(brain0, "wiki", "b", "facts")
|
|
||||||
require.NoError(t, os.MkdirAll(tdir, 0o755))
|
|
||||||
victim := filepath.Join(t.TempDir(), "victim.md")
|
|
||||||
require.NoError(t, os.WriteFile(victim, []byte("untouched\n"), 0o644))
|
|
||||||
require.NoError(t, os.Symlink(victim, filepath.Join(tdir, "t.md")))
|
|
||||||
assert.Error(t, brain.WriteTunnel(brain0, "wiki/a/facts/s.md", "wiki/b/facts/t.md"))
|
|
||||||
got, err := os.ReadFile(victim)
|
|
||||||
require.NoError(t, err)
|
|
||||||
assert.Equal(t, "untouched\n", string(got))
|
|
||||||
|
|
||||||
// The same with a link to a file elsewhere in the brain, which os.Root
|
|
||||||
// would follow.
|
|
||||||
inVictim := filepath.Join(brain0, "raw", "v.md")
|
|
||||||
require.NoError(t, os.MkdirAll(filepath.Dir(inVictim), 0o755))
|
|
||||||
require.NoError(t, os.WriteFile(inVictim, []byte("untouched\n"), 0o644))
|
|
||||||
require.NoError(t, os.Symlink(filepath.Join("..", "..", "..", "raw", "v.md"), filepath.Join(tdir, "u.md")))
|
|
||||||
assert.Error(t, brain.WriteTunnel(brain0, "wiki/a/facts/s.md", "wiki/b/facts/u.md"))
|
|
||||||
got, err = os.ReadFile(inVictim)
|
|
||||||
require.NoError(t, err)
|
|
||||||
assert.Equal(t, "untouched\n", string(got))
|
|
||||||
got, err = os.ReadFile(src)
|
|
||||||
require.NoError(t, err)
|
|
||||||
assert.Equal(t, "x\n", string(got), "the source is not touched when the target is refused")
|
|
||||||
}
|
|
||||||
|
|
||||||
// The target is swapped between a regular note and a symlink to an in-brain
|
|
||||||
// file outside wiki/ while WriteTunnel runs: the read and the write each
|
|
||||||
// refuse a symlink, so the other file is never rewritten.
|
|
||||||
func TestWriteTunnelNeverRewritesASymlinkSwappedInForTheTarget(t *testing.T) {
|
|
||||||
brain0 := t.TempDir()
|
|
||||||
src := filepath.Join(brain0, "wiki", "a", "facts", "s.md")
|
|
||||||
require.NoError(t, os.MkdirAll(filepath.Dir(src), 0o755))
|
|
||||||
require.NoError(t, os.WriteFile(src, []byte("x\n"), 0o644))
|
|
||||||
tdir := filepath.Join(brain0, "wiki", "b", "facts")
|
|
||||||
require.NoError(t, os.MkdirAll(tdir, 0o755))
|
|
||||||
victim := filepath.Join(brain0, "raw", "victim.md")
|
|
||||||
require.NoError(t, os.MkdirAll(filepath.Dir(victim), 0o755))
|
|
||||||
require.NoError(t, os.WriteFile(victim, []byte("untouched\n"), 0o644))
|
|
||||||
tgt := filepath.Join(tdir, "t.md")
|
|
||||||
require.NoError(t, os.WriteFile(tgt, []byte("t\n"), 0o644))
|
|
||||||
|
|
||||||
var stop atomic.Bool
|
|
||||||
var wg sync.WaitGroup
|
|
||||||
wg.Add(1)
|
|
||||||
go func() {
|
|
||||||
defer wg.Done()
|
|
||||||
tmp := tgt + ".swap"
|
|
||||||
for i := 0; !stop.Load(); i++ {
|
|
||||||
_ = os.Remove(tmp)
|
|
||||||
if i%2 == 0 {
|
|
||||||
_ = os.WriteFile(tmp, []byte("t\n"), 0o644)
|
|
||||||
} else {
|
|
||||||
_ = os.Symlink(filepath.Join("..", "..", "..", "raw", "victim.md"), tmp)
|
|
||||||
}
|
|
||||||
_ = os.Rename(tmp, tgt)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
ok := 0
|
|
||||||
for deadline := time.Now().Add(1500 * time.Millisecond); time.Now().Before(deadline); {
|
|
||||||
if brain.WriteTunnel(brain0, "wiki/a/facts/s.md", "wiki/b/facts/t.md") == nil {
|
|
||||||
ok++
|
|
||||||
}
|
|
||||||
_ = os.WriteFile(src, []byte("x\n"), 0o644)
|
|
||||||
}
|
|
||||||
stop.Store(true)
|
|
||||||
wg.Wait()
|
|
||||||
got, err := os.ReadFile(victim)
|
|
||||||
require.NoError(t, err)
|
|
||||||
assert.Equal(t, "untouched\n", string(got), "the swapped-in symlink's target was rewritten")
|
|
||||||
assert.Positive(t, ok, "positive control: some tunnels were written to the regular target")
|
|
||||||
}
|
|
||||||
@@ -249,41 +249,3 @@ func (b *Brain) WalkRegular(relDir string, fn func(rel string) error, onErr func
|
|||||||
return fn(p)
|
return fn(p)
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
// WriteRegular replaces the contents of an existing regular file at a
|
|
||||||
// brain-relative path. It never creates a file and never writes through a
|
|
||||||
// final symlink: os.Root.OpenFile follows an in-root final symlink even with
|
|
||||||
// O_NOFOLLOW (verified on go1.27), so the path is Lstat-ed through the root
|
|
||||||
// first and the opened handle must be that same file (os.SameFile), which
|
|
||||||
// also catches a swap between the two. O_NONBLOCK keeps a FIFO from blocking
|
|
||||||
// the open. Intermediate in-brain symlinks are still followed (os.Root keeps
|
|
||||||
// the write inside the brain; hyperguild#167).
|
|
||||||
func (b *Brain) WriteRegular(rel string, data []byte) error {
|
|
||||||
rel = filepath.ToSlash(rel)
|
|
||||||
lfi, err := b.root.Lstat(rel)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
if !lfi.Mode().IsRegular() {
|
|
||||||
return fmt.Errorf("%s: %w", rel, ErrNotRegular)
|
|
||||||
}
|
|
||||||
f, err := b.root.OpenFile(rel, os.O_WRONLY|syscall.O_NOFOLLOW|syscall.O_NONBLOCK, 0)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
defer func() { _ = f.Close() }()
|
|
||||||
fi, err := f.Stat()
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
if !fi.Mode().IsRegular() || !os.SameFile(lfi, fi) {
|
|
||||||
return fmt.Errorf("%s: changed while opening: %w", rel, ErrNotRegular)
|
|
||||||
}
|
|
||||||
if err := f.Truncate(0); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
if _, err := f.Write(data); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
return f.Close()
|
|
||||||
}
|
|
||||||
|
|||||||
@@ -263,52 +263,6 @@ func TestReadRegularNeverReturnsAnInBrainFileSwappedInBehindASymlink(t *testing.
|
|||||||
assert.Positive(t, ok, "positive control: the regular file is read")
|
assert.Positive(t, ok, "positive control: the regular file is read")
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestWriteRegularRefusesFinalSymlinksFIFOsEscapesAndMissingFilesAndRewritesRegularFiles(t *testing.T) {
|
|
||||||
dir := t.TempDir()
|
|
||||||
outside := filepath.Join(t.TempDir(), "o.md")
|
|
||||||
require.NoError(t, os.WriteFile(outside, []byte("untouched"), 0o644))
|
|
||||||
require.NoError(t, os.Symlink(outside, filepath.Join(dir, "l.md")))
|
|
||||||
require.NoError(t, syscall.Mkfifo(filepath.Join(dir, "f.md"), 0o644))
|
|
||||||
require.NoError(t, os.WriteFile(filepath.Join(dir, "n.md"), []byte("old content"), 0o644))
|
|
||||||
b, err := Open(dir)
|
|
||||||
require.NoError(t, err)
|
|
||||||
defer func() { _ = b.Close() }()
|
|
||||||
|
|
||||||
assert.Error(t, b.WriteRegular("l.md", []byte("x")))
|
|
||||||
got, _ := os.ReadFile(outside)
|
|
||||||
assert.Equal(t, "untouched", string(got))
|
|
||||||
|
|
||||||
// A symlink to another file inside the brain: os.Root would allow it,
|
|
||||||
// only O_NOFOLLOW refuses it.
|
|
||||||
require.NoError(t, os.WriteFile(filepath.Join(dir, "in.md"), []byte("inside"), 0o644))
|
|
||||||
require.NoError(t, os.Symlink("in.md", filepath.Join(dir, "il.md")))
|
|
||||||
assert.Error(t, b.WriteRegular("il.md", []byte("x")))
|
|
||||||
got, _ = os.ReadFile(filepath.Join(dir, "in.md"))
|
|
||||||
assert.Equal(t, "inside", string(got))
|
|
||||||
|
|
||||||
// A reader held open, so a non-blocking write open would succeed. The
|
|
||||||
// Lstat type check refuses the FIFO first; the handle check and
|
|
||||||
// O_NONBLOCK matter only if it is swapped in after the Lstat.
|
|
||||||
r, err := os.OpenFile(filepath.Join(dir, "f.md"), os.O_RDONLY|syscall.O_NONBLOCK, 0)
|
|
||||||
require.NoError(t, err)
|
|
||||||
defer func() { _ = r.Close() }()
|
|
||||||
done := make(chan error, 1)
|
|
||||||
go func() { done <- b.WriteRegular("f.md", []byte("x")) }()
|
|
||||||
select {
|
|
||||||
case err := <-done:
|
|
||||||
assert.Error(t, err)
|
|
||||||
case <-time.After(3 * time.Second):
|
|
||||||
t.Fatal("WriteRegular hung on a FIFO")
|
|
||||||
}
|
|
||||||
|
|
||||||
assert.Error(t, b.WriteRegular("../"+filepath.Base(filepath.Dir(outside))+"/o.md", []byte("x")))
|
|
||||||
assert.Error(t, b.WriteRegular("missing.md", []byte("x")), "does not create files")
|
|
||||||
|
|
||||||
require.NoError(t, b.WriteRegular("n.md", []byte("new")))
|
|
||||||
got, _ = os.ReadFile(filepath.Join(dir, "n.md"))
|
|
||||||
assert.Equal(t, "new", string(got), "truncated and rewritten")
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestReadDirDirsListsOnlyRealSubdirsAndRefusesSymlinkedComponents(t *testing.T) {
|
func TestReadDirDirsListsOnlyRealSubdirsAndRefusesSymlinkedComponents(t *testing.T) {
|
||||||
dir := t.TempDir()
|
dir := t.TempDir()
|
||||||
require.NoError(t, os.MkdirAll(filepath.Join(dir, "a", "real"), 0o755))
|
require.NoError(t, os.MkdirAll(filepath.Join(dir, "a", "real"), 0o755))
|
||||||
|
|||||||
@@ -16,10 +16,9 @@
|
|||||||
package classification
|
package classification
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"errors"
|
|
||||||
"fmt"
|
"fmt"
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
|
"os"
|
||||||
"io/fs"
|
"path/filepath"
|
||||||
"strings"
|
"strings"
|
||||||
|
|
||||||
"gopkg.in/yaml.v3"
|
"gopkg.in/yaml.v3"
|
||||||
@@ -111,20 +110,9 @@ type rawConfig struct {
|
|||||||
func Load(brainDir string) (*Config, error) {
|
func Load(brainDir string) (*Config, error) {
|
||||||
cfg := &Config{wings: map[string]Level{}, repos: map[string]Level{}}
|
cfg := &Config{wings: map[string]Level{}, repos: map[string]Level{}}
|
||||||
|
|
||||||
// Read through the brain root, regular files only (hyperguild#157): a
|
data, err := os.ReadFile(filepath.Join(brainDir, "classification.yaml"))
|
||||||
// symlinked or special file is not trusted as the config, which is a hard
|
|
||||||
// error like a malformed one.
|
|
||||||
bfs, err := brainfs.Open(brainDir)
|
|
||||||
if errors.Is(err, fs.ErrNotExist) {
|
|
||||||
return cfg, nil
|
|
||||||
}
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("open brain: %w", err)
|
if os.IsNotExist(err) {
|
||||||
}
|
|
||||||
defer func() { _ = bfs.Close() }()
|
|
||||||
data, err := bfs.ReadRegular("classification.yaml")
|
|
||||||
if err != nil {
|
|
||||||
if errors.Is(err, fs.ErrNotExist) {
|
|
||||||
return cfg, nil
|
return cfg, nil
|
||||||
}
|
}
|
||||||
return nil, fmt.Errorf("read classification.yaml: %w", err)
|
return nil, fmt.Errorf("read classification.yaml: %w", err)
|
||||||
|
|||||||
@@ -1,46 +0,0 @@
|
|||||||
package classification
|
|
||||||
|
|
||||||
import (
|
|
||||||
"os"
|
|
||||||
"path/filepath"
|
|
||||||
"syscall"
|
|
||||||
"testing"
|
|
||||||
"time"
|
|
||||||
|
|
||||||
"github.com/stretchr/testify/assert"
|
|
||||||
"github.com/stretchr/testify/require"
|
|
||||||
)
|
|
||||||
|
|
||||||
// hyperguild#157 slice 6: Load read classification.yaml with os.ReadFile. It
|
|
||||||
// sets the I1 sovereignty levels; a symlinked file must not be trusted as the
|
|
||||||
// config (a hard error, as for a malformed file), and a FIFO must not hang
|
|
||||||
// startup.
|
|
||||||
func TestLoadRefusesASymlinkedOrFIFOClassificationFile(t *testing.T) {
|
|
||||||
ok := t.TempDir()
|
|
||||||
require.NoError(t, os.WriteFile(filepath.Join(ok, "classification.yaml"), []byte("wings:\n w: confidential\n"), 0o644))
|
|
||||||
_, err := Load(ok)
|
|
||||||
require.NoError(t, err, "positive control")
|
|
||||||
|
|
||||||
linked := t.TempDir()
|
|
||||||
require.NoError(t, os.WriteFile(filepath.Join(linked, "real.yaml"), []byte("wings:\n w: public\n"), 0o644))
|
|
||||||
require.NoError(t, os.Symlink("real.yaml", filepath.Join(linked, "classification.yaml")))
|
|
||||||
_, err = Load(linked)
|
|
||||||
assert.Error(t, err, "an in-brain symlink is not trusted as the config")
|
|
||||||
|
|
||||||
fifo := t.TempDir()
|
|
||||||
require.NoError(t, syscall.Mkfifo(filepath.Join(fifo, "classification.yaml"), 0o644))
|
|
||||||
w, err := os.OpenFile(filepath.Join(fifo, "classification.yaml"), os.O_RDWR, 0)
|
|
||||||
require.NoError(t, err)
|
|
||||||
defer func() { _ = w.Close() }()
|
|
||||||
done := make(chan error, 1)
|
|
||||||
go func() { _, err := Load(fifo); done <- err }()
|
|
||||||
select {
|
|
||||||
case err := <-done:
|
|
||||||
assert.Error(t, err)
|
|
||||||
case <-time.After(3 * time.Second):
|
|
||||||
t.Fatal("Load hung on a FIFO")
|
|
||||||
}
|
|
||||||
|
|
||||||
_, err = Load(t.TempDir())
|
|
||||||
assert.NoError(t, err, "an absent file is still not an error")
|
|
||||||
}
|
|
||||||
@@ -3,10 +3,7 @@ package pipeline
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"errors"
|
|
||||||
"fmt"
|
"fmt"
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
|
|
||||||
"io/fs"
|
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"strings"
|
"strings"
|
||||||
@@ -26,16 +23,10 @@ func BackfillRefs(ctx context.Context, brainDir string) (int, error) {
|
|||||||
return 0, fmt.Errorf("load inventory: %w", err)
|
return 0, fmt.Errorf("load inventory: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Listed and read through the brain root, regular files only
|
sourcesDir := filepath.Join(brainDir, "wiki", "sources")
|
||||||
// (hyperguild#157).
|
entries, err := os.ReadDir(sourcesDir)
|
||||||
bfs, err := brainfs.Open(brainDir)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return 0, fmt.Errorf("open brain: %w", err)
|
if os.IsNotExist(err) {
|
||||||
}
|
|
||||||
defer func() { _ = bfs.Close() }()
|
|
||||||
names, err := bfs.ReadDirRegular("wiki/sources")
|
|
||||||
if err != nil {
|
|
||||||
if errors.Is(err, fs.ErrNotExist) {
|
|
||||||
return 0, nil
|
return 0, nil
|
||||||
}
|
}
|
||||||
return 0, fmt.Errorf("read sources dir: %w", err)
|
return 0, fmt.Errorf("read sources dir: %w", err)
|
||||||
@@ -46,20 +37,20 @@ func BackfillRefs(ctx context.Context, brainDir string) (int, error) {
|
|||||||
// get both refs merged before a single write.
|
// get both refs merged before a single write.
|
||||||
pending := make(map[string]wiki.Page)
|
pending := make(map[string]wiki.Page)
|
||||||
|
|
||||||
for _, name := range names {
|
for _, e := range entries {
|
||||||
if ctx.Err() != nil {
|
if ctx.Err() != nil {
|
||||||
return 0, ctx.Err()
|
return 0, ctx.Err()
|
||||||
}
|
}
|
||||||
if !strings.HasSuffix(name, ".md") {
|
if e.IsDir() || !strings.HasSuffix(e.Name(), ".md") {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
b, err := bfs.ReadRegular("wiki/sources/" + name)
|
b, err := os.ReadFile(filepath.Join(sourcesDir, e.Name()))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
sourceContent := string(b)
|
sourceContent := string(b)
|
||||||
sourceSlug := strings.TrimSuffix(name, ".md")
|
sourceSlug := strings.TrimSuffix(e.Name(), ".md")
|
||||||
sourceTitle := extractTitle(sourceContent)
|
sourceTitle := extractTitle(sourceContent)
|
||||||
if sourceTitle == "" {
|
if sourceTitle == "" {
|
||||||
sourceTitle = sourceSlug
|
sourceTitle = sourceSlug
|
||||||
@@ -79,7 +70,7 @@ func BackfillRefs(ctx context.Context, brainDir string) (int, error) {
|
|||||||
// Start from already-accumulated version if we've seen this page.
|
// Start from already-accumulated version if we've seen this page.
|
||||||
page, seen := pending[relPath]
|
page, seen := pending[relPath]
|
||||||
if !seen {
|
if !seen {
|
||||||
raw, err := bfs.ReadRegular(relPath)
|
raw, err := os.ReadFile(filepath.Join(brainDir, filepath.FromSlash(relPath)))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -5,10 +5,7 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"crypto/sha256"
|
"crypto/sha256"
|
||||||
"encoding/hex"
|
"encoding/hex"
|
||||||
"errors"
|
|
||||||
"fmt"
|
"fmt"
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
|
|
||||||
"io/fs"
|
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"strings"
|
"strings"
|
||||||
@@ -132,7 +129,7 @@ func buildAndWrite(ctx context.Context, rawPages []RawPage, sourceSlug, date, br
|
|||||||
if err := os.MkdirAll(filepath.Dir(dest), 0o755); err != nil {
|
if err := os.MkdirAll(filepath.Dir(dest), 0o755); err != nil {
|
||||||
return Result{}, fmt.Errorf("mkdir for %s: %w", page.Path, err)
|
return Result{}, fmt.Errorf("mkdir for %s: %w", page.Path, err)
|
||||||
}
|
}
|
||||||
content, conflicts, err := reviseOrCreate(brainDir, page)
|
content, conflicts, err := reviseOrCreate(dest, page)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return Result{}, err
|
return Result{}, err
|
||||||
}
|
}
|
||||||
@@ -191,27 +188,14 @@ func hashBody(content string) string {
|
|||||||
|
|
||||||
// readSourcePage returns the full on-disk content of the source page at
|
// readSourcePage returns the full on-disk content of the source page at
|
||||||
// wiki/sources/<sourceSlug>.md, or "" when it doesn't exist yet.
|
// wiki/sources/<sourceSlug>.md, or "" when it doesn't exist yet.
|
||||||
// It reads through the brain root, regular files only (hyperguild#157): the
|
|
||||||
// content goes into the LLM prompt.
|
|
||||||
func readSourcePage(brainDir, sourceSlug string) string {
|
func readSourcePage(brainDir, sourceSlug string) string {
|
||||||
raw, err := readBrainFile(brainDir, "wiki/sources/"+sourceSlug+".md")
|
raw, err := os.ReadFile(filepath.Join(brainDir, "wiki", "sources", sourceSlug+".md"))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return ""
|
return ""
|
||||||
}
|
}
|
||||||
return string(raw)
|
return string(raw)
|
||||||
}
|
}
|
||||||
|
|
||||||
// readBrainFile reads one brain-relative file through the brain root,
|
|
||||||
// regular files only (hyperguild#157).
|
|
||||||
func readBrainFile(brainDir, rel string) ([]byte, error) {
|
|
||||||
bfs, err := brainfs.Open(brainDir)
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
defer func() { _ = bfs.Close() }()
|
|
||||||
return bfs.ReadRegular(rel)
|
|
||||||
}
|
|
||||||
|
|
||||||
// storedSourceHash reads source_hash from the source page a previous ingest
|
// storedSourceHash reads source_hash from the source page a previous ingest
|
||||||
// wrote, or "" when there is no such page or no such field.
|
// wrote, or "" when there is no such page or no such field.
|
||||||
//
|
//
|
||||||
@@ -246,10 +230,10 @@ func storedSourceHash(brainDir, sourceSlug string) string {
|
|||||||
// its first-ingest date on a revised page. Left deliberately — the alternative
|
// its first-ingest date on a revised page. Left deliberately — the alternative
|
||||||
// is this function editing frontmatter it does not own, and no lint reads that
|
// is this function editing frontmatter it does not own, and no lint reads that
|
||||||
// field today.
|
// field today.
|
||||||
func reviseOrCreate(brainDir string, page wiki.Page) (string, []string, error) {
|
func reviseOrCreate(dest string, page wiki.Page) (string, []string, error) {
|
||||||
existing, err := readBrainFile(brainDir, page.Path)
|
existing, err := os.ReadFile(dest)
|
||||||
switch {
|
switch {
|
||||||
case errors.Is(err, fs.ErrNotExist):
|
case os.IsNotExist(err):
|
||||||
return page.Content, nil, nil
|
return page.Content, nil, nil
|
||||||
case err != nil:
|
case err != nil:
|
||||||
return "", nil, fmt.Errorf("read existing %s: %w", page.Path, err)
|
return "", nil, fmt.Errorf("read existing %s: %w", page.Path, err)
|
||||||
@@ -283,7 +267,7 @@ See brain/schema.md for the full schema.
|
|||||||
`
|
`
|
||||||
|
|
||||||
func loadSchema(brainDir string) string {
|
func loadSchema(brainDir string) string {
|
||||||
b, err := readBrainFile(brainDir, "schema.md")
|
b, err := os.ReadFile(filepath.Join(brainDir, "schema.md"))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return defaultSchema
|
return defaultSchema
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -2,6 +2,7 @@
|
|||||||
package pipeline
|
package pipeline
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"regexp"
|
"regexp"
|
||||||
"strings"
|
"strings"
|
||||||
@@ -52,7 +53,8 @@ func injectSourceRefs(pages []wiki.Page, inventory map[wiki.PageType][]wiki.Entr
|
|||||||
if !ok {
|
if !ok {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
b, err := readBrainFile(brainDir, "wiki/"+string(pt)+"/"+slug+".md")
|
diskPath := filepath.Join(brainDir, "wiki", string(pt), slug+".md")
|
||||||
|
b, err := os.ReadFile(diskPath)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,167 +0,0 @@
|
|||||||
package pipeline
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
"fmt"
|
|
||||||
"io"
|
|
||||||
"os"
|
|
||||||
"path/filepath"
|
|
||||||
"syscall"
|
|
||||||
"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/wiki"
|
|
||||||
)
|
|
||||||
|
|
||||||
// hyperguild#157 slice 6: the pipeline read the source page, schema.md, the
|
|
||||||
// pages it revises and the pages it adds source refs to with os.ReadFile, so
|
|
||||||
// a symlinked file's content reached the LLM prompt and the written pages,
|
|
||||||
// and a FIFO hung the ingest.
|
|
||||||
|
|
||||||
func mkdirs(t *testing.T, brain string, dirs ...string) {
|
|
||||||
t.Helper()
|
|
||||||
for _, d := range dirs {
|
|
||||||
require.NoError(t, os.MkdirAll(filepath.Join(brain, filepath.FromSlash(d)), 0o755))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestReadSourcePageAndSchemaRefuseSymlinksAndSurviveSwapRaces(t *testing.T) {
|
|
||||||
brain := t.TempDir()
|
|
||||||
mkdirs(t, brain, "wiki/sources")
|
|
||||||
inBrain := brainfstest.WriteSentinelAt(t, filepath.Join(brain, "raw", "s.md"))
|
|
||||||
require.NoError(t, os.WriteFile(filepath.Join(brain, "wiki", "sources", "ok.md"), []byte("zqinside source"), 0o644))
|
|
||||||
require.NoError(t, os.WriteFile(filepath.Join(brain, "schema.md"), []byte("zqinside schema"), 0o644))
|
|
||||||
assert.Equal(t, "zqinside source", readSourcePage(brain, "ok"), "positive control")
|
|
||||||
assert.Equal(t, "zqinside schema", loadSchema(brain), "positive control")
|
|
||||||
|
|
||||||
require.NoError(t, os.Symlink(brainfstest.Outside(t), filepath.Join(brain, "wiki", "sources", "out.md")))
|
|
||||||
require.NoError(t, os.Symlink(filepath.Join("..", "..", "raw", "s.md"), filepath.Join(brain, "wiki", "sources", "in.md")))
|
|
||||||
assert.NotContains(t, readSourcePage(brain, "out"), brainfstest.Sentinel)
|
|
||||||
assert.NotContains(t, readSourcePage(brain, "in"), brainfstest.Sentinel)
|
|
||||||
assert.NotContains(t, storedSourceHash(brain, "in"), brainfstest.Sentinel)
|
|
||||||
|
|
||||||
src := filepath.Join(brain, "wiki", "sources", "r.md")
|
|
||||||
read := func() string { return readSourcePage(brain, "r") }
|
|
||||||
brainfstest.AssertNoLeak(t, src, "zqinside", 1500*time.Millisecond, read)
|
|
||||||
brainfstest.AssertNoLeakFrom(t, src, "zqinside", inBrain, 1500*time.Millisecond, read)
|
|
||||||
brainfstest.AssertNoHang(t, src, "zqinside", 1500*time.Millisecond, 6*time.Second, func() { _ = read() })
|
|
||||||
|
|
||||||
schema := filepath.Join(brain, "schema.md")
|
|
||||||
rs := func() string { return loadSchema(brain) }
|
|
||||||
brainfstest.AssertNoLeakFrom(t, schema, "zqinside schema", inBrain, 1500*time.Millisecond, rs)
|
|
||||||
brainfstest.AssertNoHang(t, schema, "zqinside schema", 1500*time.Millisecond, 6*time.Second, func() { _ = rs() })
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestReviseOrCreateRefusesASymlinkedExistingPage(t *testing.T) {
|
|
||||||
brain := t.TempDir()
|
|
||||||
mkdirs(t, brain, "wiki/concepts")
|
|
||||||
brainfstest.WriteSentinelAt(t, filepath.Join(brain, "raw", "s.md"))
|
|
||||||
require.NoError(t, os.Symlink(filepath.Join("..", "..", "raw", "s.md"), filepath.Join(brain, "wiki", "concepts", "c.md")))
|
|
||||||
page := wiki.Page{Path: "wiki/concepts/c.md", Content: "---\ntitle: C\n---\nnew\n"}
|
|
||||||
out, _, err := reviseOrCreate(brain, page)
|
|
||||||
assert.Error(t, err)
|
|
||||||
assert.NotContains(t, out, brainfstest.Sentinel)
|
|
||||||
|
|
||||||
require.NoError(t, os.WriteFile(filepath.Join(brain, "wiki", "concepts", "d.md"), []byte("---\ntitle: D\n---\nzqold\n"), 0o644))
|
|
||||||
out, _, err = reviseOrCreate(brain, wiki.Page{Path: "wiki/concepts/d.md", Content: "---\ntitle: D\n---\nnew\n"})
|
|
||||||
require.NoError(t, err)
|
|
||||||
assert.Contains(t, out, "zqold", "positive control: the existing page is read")
|
|
||||||
out, _, err = reviseOrCreate(brain, wiki.Page{Path: "wiki/concepts/new.md", Content: "fresh"})
|
|
||||||
require.NoError(t, err)
|
|
||||||
assert.Equal(t, "fresh", out, "positive control: a missing page is created")
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestInjectSourceRefsDoesNotReadASymlinkedPage(t *testing.T) {
|
|
||||||
brain := t.TempDir()
|
|
||||||
mkdirs(t, brain, "wiki/concepts")
|
|
||||||
brainfstest.WriteSentinelAt(t, filepath.Join(brain, "raw", "s.md"))
|
|
||||||
require.NoError(t, os.Symlink(filepath.Join("..", "..", "raw", "s.md"), filepath.Join(brain, "wiki", "concepts", "c.md")))
|
|
||||||
require.NoError(t, os.WriteFile(filepath.Join(brain, "wiki", "concepts", "d.md"), []byte("zqinside d"), 0o644))
|
|
||||||
inv := map[wiki.PageType][]wiki.Entry{wiki.PageTypeConcept: {
|
|
||||||
{Slug: "c", Type: wiki.PageTypeConcept}, {Slug: "d", Type: wiki.PageTypeConcept},
|
|
||||||
}}
|
|
||||||
src := wiki.Page{Path: "wiki/sources/src.md", Content: "---\ntitle: Src\n---\nsee [[c|C]] and [[d|D]]\n"}
|
|
||||||
out := injectSourceRefs([]wiki.Page{src}, inv, brain)
|
|
||||||
got := fmt.Sprintf("%+v", out)
|
|
||||||
assert.Contains(t, got, "zqinside d", "positive control")
|
|
||||||
assert.NotContains(t, got, brainfstest.Sentinel)
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestBackfillRefsSkipsSymlinkedSourcesAndPages(t *testing.T) {
|
|
||||||
brain := t.TempDir()
|
|
||||||
mkdirs(t, brain, "wiki/sources", "wiki/concepts", "raw")
|
|
||||||
require.NoError(t, os.WriteFile(filepath.Join(brain, "wiki", "concepts", "real.md"), []byte("---\ntitle: R\n---\nbody\n"), 0o644))
|
|
||||||
require.NoError(t, os.WriteFile(filepath.Join(brain, "wiki", "sources", "ok.md"), []byte("---\ntitle: zqinside\n---\n[[real|R]]\n"), 0o644))
|
|
||||||
// A source elsewhere in the brain whose title is the sentinel and which
|
|
||||||
// links the real concept with the [[slug|title]] form the pipeline follows.
|
|
||||||
require.NoError(t, os.WriteFile(filepath.Join(brain, "raw", "src.md"),
|
|
||||||
[]byte("---\ntitle: "+brainfstest.Sentinel+"\n---\n[[real|R]]\n"), 0o644))
|
|
||||||
require.NoError(t, os.Symlink(filepath.Join("..", "..", "raw", "src.md"), filepath.Join(brain, "wiki", "sources", "in.md")))
|
|
||||||
// A concept page linked by the regular source that is itself a symlink.
|
|
||||||
require.NoError(t, os.WriteFile(filepath.Join(brain, "wiki", "sources", "ok2.md"), []byte("---\ntitle: Two\n---\n[[lnk|L]]\n"), 0o644))
|
|
||||||
brainfstest.WriteSentinelAt(t, filepath.Join(brain, "raw", "page.md"))
|
|
||||||
require.NoError(t, os.Symlink(filepath.Join("..", "..", "raw", "page.md"), filepath.Join(brain, "wiki", "concepts", "lnk.md")))
|
|
||||||
|
|
||||||
_, err := BackfillRefs(context.Background(), brain)
|
|
||||||
require.NoError(t, err)
|
|
||||||
b, _ := os.ReadFile(filepath.Join(brain, "wiki", "concepts", "real.md"))
|
|
||||||
assert.Contains(t, string(b), "zqinside", "positive control: the regular source's ref is added")
|
|
||||||
assert.NotContains(t, string(b), brainfstest.Sentinel, "a symlinked source's title was written as a ref")
|
|
||||||
raw, _ := os.ReadFile(filepath.Join(brain, "raw", "page.md"))
|
|
||||||
assert.NotContains(t, string(raw), "[[ok2", "the symlinked concept page's target was rewritten")
|
|
||||||
}
|
|
||||||
|
|
||||||
// A source swapped between a regular file and a symlink to an in-brain source
|
|
||||||
// whose title is the sentinel: the read refuses the swapped-in link, so the
|
|
||||||
// sentinel never lands in the concept page as a ref.
|
|
||||||
func TestBackfillRefsDoesNotWriteARefFromASourceSwappedInBehindASymlink(t *testing.T) {
|
|
||||||
brain := t.TempDir()
|
|
||||||
mkdirs(t, brain, "wiki/sources", "wiki/concepts", "raw")
|
|
||||||
realPage := filepath.Join(brain, "wiki", "concepts", "real.md")
|
|
||||||
target := filepath.Join(brain, "raw", "src.md")
|
|
||||||
require.NoError(t, os.WriteFile(target, []byte("---\ntitle: "+brainfstest.Sentinel+"\n---\n[[real|R]]\n"), 0o644))
|
|
||||||
src := filepath.Join(brain, "wiki", "sources", "r.md")
|
|
||||||
inside := "---\ntitle: zqinside\n---\n[[real|R]]\n"
|
|
||||||
read := func() string {
|
|
||||||
_ = os.WriteFile(realPage, []byte("---\ntitle: R\n---\nbody\n"), 0o644)
|
|
||||||
_, _ = BackfillRefs(context.Background(), brain)
|
|
||||||
b, _ := os.ReadFile(realPage)
|
|
||||||
return string(b)
|
|
||||||
}
|
|
||||||
require.NoError(t, os.WriteFile(src, []byte(inside), 0o644))
|
|
||||||
require.Contains(t, read(), "zqinside", "positive control")
|
|
||||||
brainfstest.AssertNoLeakFrom(t, src, inside, target, 1500*time.Millisecond, read)
|
|
||||||
}
|
|
||||||
|
|
||||||
// A concept page that is regular when the inventory lists it, then swapped
|
|
||||||
// for a symlink to an in-brain file before BackfillRefs reads it: only the
|
|
||||||
// page read's own check refuses it (a static symlink never gets this far,
|
|
||||||
// because LoadInventory does not list it).
|
|
||||||
func TestBackfillRefsDoesNotCopyAPageSwappedInBehindASymlink(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")
|
|
||||||
inside := "---\ntitle: C\n---\nzqinside\n"
|
|
||||||
read := func() string {
|
|
||||||
// The write path still follows a swapped-in link (#158) and can
|
|
||||||
// overwrite the target, so restore it each round.
|
|
||||||
_ = os.WriteFile(target, []byte("---\ntitle: X\n---\n"+brainfstest.Sentinel+"\n"), 0o644)
|
|
||||||
_, _ = BackfillRefs(context.Background(), brain)
|
|
||||||
f, err := os.OpenFile(page, os.O_RDONLY|syscall.O_NOFOLLOW|syscall.O_NONBLOCK, 0)
|
|
||||||
if err != nil {
|
|
||||||
return ""
|
|
||||||
}
|
|
||||||
defer func() { _ = f.Close() }()
|
|
||||||
b, _ := io.ReadAll(f)
|
|
||||||
return string(b)
|
|
||||||
}
|
|
||||||
require.NoError(t, os.WriteFile(page, []byte(inside), 0o644))
|
|
||||||
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)
|
|
||||||
}
|
|
||||||
@@ -1,92 +0,0 @@
|
|||||||
package session
|
|
||||||
|
|
||||||
import (
|
|
||||||
"errors"
|
|
||||||
"os"
|
|
||||||
"path/filepath"
|
|
||||||
"syscall"
|
|
||||||
"testing"
|
|
||||||
"time"
|
|
||||||
|
|
||||||
"github.com/stretchr/testify/assert"
|
|
||||||
"github.com/stretchr/testify/require"
|
|
||||||
)
|
|
||||||
|
|
||||||
// session_log passed the caller's session_id straight into
|
|
||||||
// filepath.Join(sessionsDir, id+".jsonl") and opened it O_CREATE|O_APPEND:
|
|
||||||
// path segments in the id created or appended files anywhere the pod could
|
|
||||||
// write. The id is now a plain name, and the open does not follow a final
|
|
||||||
// symlink.
|
|
||||||
func TestAppendRefusesSessionIDsThatAreNotPlainNames(t *testing.T) {
|
|
||||||
root := t.TempDir()
|
|
||||||
dir := filepath.Join(root, "sessions")
|
|
||||||
for _, id := range []string{"../x", "a/b", "/abs", ".", "..", ".hidden", "", "a\\b", "a\x00b"} {
|
|
||||||
assert.Error(t, Append(dir, id, Entry{SessionID: id}), "%q", id)
|
|
||||||
}
|
|
||||||
ents, _ := os.ReadDir(root)
|
|
||||||
for _, e := range ents {
|
|
||||||
assert.Equal(t, "sessions", e.Name(), "nothing created beside the sessions dir")
|
|
||||||
}
|
|
||||||
for _, id := range []string{"1557b2e1-bf0d-5c4f-bbe3-0754d087e356", "_routing", "2026-05-05-infra.plan", "0"} {
|
|
||||||
require.NoError(t, Append(dir, id, Entry{SessionID: id}), "positive control: %q", id)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestAppendDoesNotWriteThroughASymlinkedLog(t *testing.T) {
|
|
||||||
root := t.TempDir()
|
|
||||||
dir := filepath.Join(root, "sessions")
|
|
||||||
require.NoError(t, os.MkdirAll(dir, 0o755))
|
|
||||||
victim := filepath.Join(root, "victim.md")
|
|
||||||
require.NoError(t, os.WriteFile(victim, []byte("untouched\n"), 0o644))
|
|
||||||
require.NoError(t, os.Symlink(filepath.Join("..", "victim.md"), filepath.Join(dir, "s.jsonl")))
|
|
||||||
assert.Error(t, Append(dir, "s", Entry{SessionID: "s"}))
|
|
||||||
got, _ := os.ReadFile(victim)
|
|
||||||
assert.Equal(t, "untouched\n", string(got))
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestReadRefusesSessionIDsThatAreNotPlainNames(t *testing.T) {
|
|
||||||
root := t.TempDir()
|
|
||||||
require.NoError(t, os.WriteFile(filepath.Join(root, "x.jsonl"), []byte(`{"session_id":"x"}`+"\n"), 0o644))
|
|
||||||
_, err := Read(filepath.Join(root, "sessions"), "../x")
|
|
||||||
assert.Error(t, err)
|
|
||||||
}
|
|
||||||
|
|
||||||
// A name that starts plainly and then climbs out with ".." is exactly what the
|
|
||||||
// ban on "/" in the id's tail stops: filepath.Join cleans "a/../../x" to a
|
|
||||||
// path beside the sessions dir without needing sessions/a to exist. The
|
|
||||||
// assertion is on the refusal itself (ErrInvalidID), not on any error.
|
|
||||||
func TestAppendRefusesAnIDThatClimbsOutAfterAPlainPrefix(t *testing.T) {
|
|
||||||
root := t.TempDir()
|
|
||||||
dir := filepath.Join(root, "sessions")
|
|
||||||
require.NoError(t, os.MkdirAll(dir, 0o755))
|
|
||||||
for _, id := range []string{"a/../../escape", "a/../../../escape", "a/../b"} {
|
|
||||||
err := Append(dir, id, Entry{SessionID: id})
|
|
||||||
assert.ErrorIs(t, err, ErrInvalidID, "%q", id)
|
|
||||||
}
|
|
||||||
_, err := os.Lstat(filepath.Join(root, "escape.jsonl"))
|
|
||||||
assert.True(t, errors.Is(err, os.ErrNotExist), "nothing created beside the sessions dir")
|
|
||||||
_, err = os.Lstat(filepath.Join(dir, "b.jsonl"))
|
|
||||||
assert.True(t, errors.Is(err, os.ErrNotExist), "no in-dir alias created either")
|
|
||||||
}
|
|
||||||
|
|
||||||
// Append does no Lstat type check, so O_NONBLOCK is the only thing that keeps
|
|
||||||
// a FIFO named <id>.jsonl with no reader from blocking session_log forever.
|
|
||||||
func TestAppendReturnsPromptlyOnAFIFOLogWithNoReader(t *testing.T) {
|
|
||||||
dir := filepath.Join(t.TempDir(), "sessions")
|
|
||||||
require.NoError(t, os.MkdirAll(dir, 0o755))
|
|
||||||
fifo := filepath.Join(dir, "s.jsonl")
|
|
||||||
require.NoError(t, syscall.Mkfifo(fifo, 0o644))
|
|
||||||
done := make(chan error, 1)
|
|
||||||
go func() { done <- Append(dir, "s", Entry{SessionID: "s"}) }()
|
|
||||||
select {
|
|
||||||
case err := <-done:
|
|
||||||
assert.Error(t, err, "a FIFO with no reader is refused, not written")
|
|
||||||
case <-time.After(3 * time.Second):
|
|
||||||
// Unblock the stuck open so the goroutine and TempDir cleanup finish.
|
|
||||||
if r, err := os.OpenFile(fifo, os.O_RDONLY|syscall.O_NONBLOCK, 0); err == nil {
|
|
||||||
defer r.Close() //nolint:errcheck
|
|
||||||
}
|
|
||||||
<-done
|
|
||||||
t.Fatal("Append blocked on a FIFO log with no reader")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -9,8 +9,6 @@ import (
|
|||||||
"io/fs"
|
"io/fs"
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"regexp"
|
|
||||||
"syscall"
|
|
||||||
"time"
|
"time"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -44,32 +42,13 @@ type Attempt struct {
|
|||||||
RunnerOutput string `json:"runner_output,omitempty"`
|
RunnerOutput string `json:"runner_output,omitempty"`
|
||||||
}
|
}
|
||||||
|
|
||||||
// idPattern is a plain file name: session_log takes the id from the caller,
|
|
||||||
// and a path segment in it created or appended files anywhere the pod could
|
|
||||||
// write (hyperguild#169).
|
|
||||||
var idPattern = regexp.MustCompile(`^[A-Za-z0-9_][A-Za-z0-9._-]{0,127}$`)
|
|
||||||
|
|
||||||
// ErrInvalidID is returned when a session id is not a plain file name.
|
|
||||||
var ErrInvalidID = errors.New("session id is not a plain name")
|
|
||||||
|
|
||||||
func checkID(sessionID string) error {
|
|
||||||
if !idPattern.MatchString(sessionID) {
|
|
||||||
return fmt.Errorf("%w: %q", ErrInvalidID, sessionID)
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// Append writes entry as a single JSON line to sessionsDir/{sessionID}.jsonl.
|
// Append writes entry as a single JSON line to sessionsDir/{sessionID}.jsonl.
|
||||||
// The open does not follow a final symlink.
|
|
||||||
func Append(sessionsDir, sessionID string, entry Entry) error {
|
func Append(sessionsDir, sessionID string, entry Entry) error {
|
||||||
if err := checkID(sessionID); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
if err := os.MkdirAll(sessionsDir, 0o755); err != nil {
|
if err := os.MkdirAll(sessionsDir, 0o755); err != nil {
|
||||||
return fmt.Errorf("create sessions dir: %w", err)
|
return fmt.Errorf("create sessions dir: %w", err)
|
||||||
}
|
}
|
||||||
path := filepath.Join(sessionsDir, sessionID+".jsonl")
|
path := filepath.Join(sessionsDir, sessionID+".jsonl")
|
||||||
f, err := os.OpenFile(path, os.O_CREATE|os.O_APPEND|os.O_WRONLY|syscall.O_NOFOLLOW|syscall.O_NONBLOCK, 0o644)
|
f, err := os.OpenFile(path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o644)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("open session log: %w", err)
|
return fmt.Errorf("open session log: %w", err)
|
||||||
}
|
}
|
||||||
@@ -91,9 +70,6 @@ func Append(sessionsDir, sessionID string, entry Entry) error {
|
|||||||
|
|
||||||
// Read returns all entries for sessionID. Returns empty slice if no log exists.
|
// Read returns all entries for sessionID. Returns empty slice if no log exists.
|
||||||
func Read(sessionsDir, sessionID string) ([]Entry, error) {
|
func Read(sessionsDir, sessionID string) ([]Entry, error) {
|
||||||
if err := checkID(sessionID); err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
path := filepath.Join(sessionsDir, sessionID+".jsonl")
|
path := filepath.Join(sessionsDir, sessionID+".jsonl")
|
||||||
f, err := os.Open(path)
|
f, err := os.Open(path)
|
||||||
if errors.Is(err, fs.ErrNotExist) {
|
if errors.Is(err, fs.ErrNotExist) {
|
||||||
|
|||||||
@@ -18,14 +18,9 @@ import (
|
|||||||
"sort"
|
"sort"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
|
||||||
"sync/atomic"
|
|
||||||
"time"
|
"time"
|
||||||
)
|
)
|
||||||
|
|
||||||
// shutdownFlushTimeout bounds the final flush when Watch stops.
|
|
||||||
var shutdownFlushTimeout = 5 * time.Second
|
|
||||||
|
|
||||||
// DefaultInterval is used when Watch is given a non-positive interval.
|
// DefaultInterval is used when Watch is given a non-positive interval.
|
||||||
const DefaultInterval = 30 * time.Second
|
const DefaultInterval = 30 * time.Second
|
||||||
|
|
||||||
@@ -138,32 +133,13 @@ func Scan(brainDir string) (Result, error) {
|
|||||||
// queue until Notify succeeds (see alertQueue): an outage delays them, and
|
// queue until Notify succeeds (see alertQueue): an outage delays them, and
|
||||||
// any that do not fit the bounded queue, or are undeliverable, are counted in
|
// any that do not fit the bounded queue, or are undeliverable, are counted in
|
||||||
// a summary that is never dropped, so every incident is delivered or counted.
|
// a summary that is never dropped, so every incident is delivered or counted.
|
||||||
//
|
|
||||||
// Delivery runs in its own goroutine (hyperguild#164): a hung notifier, with
|
|
||||||
// a full queue, blocked one flush for several notify timeouts, and while the
|
|
||||||
// scan waited on it a planted link stayed readable for several intervals.
|
|
||||||
// The scan hands alerts over without blocking; any that do not fit are
|
|
||||||
// counted in the summary. When ctx ends, Watch returns after a final flush
|
|
||||||
// of what was handed over (bounded by shutdownFlushTimeout); alerts that
|
|
||||||
// flush cannot deliver are counted in one error log line instead. That
|
|
||||||
// shutdown path only runs for a caller that cancels ctx: cmd/server passes
|
|
||||||
// context.Background(), so in production the flush on handoff is what keeps
|
|
||||||
// alerts from waiting on a restart.
|
|
||||||
func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notifier) {
|
func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notifier) {
|
||||||
if interval <= 0 {
|
if interval <= 0 {
|
||||||
interval = DefaultInterval
|
interval = DefaultInterval
|
||||||
}
|
}
|
||||||
t := time.NewTicker(interval)
|
t := time.NewTicker(interval)
|
||||||
defer t.Stop()
|
defer t.Stop()
|
||||||
alerts := make(chan string, maxPending)
|
q := &alertQueue{}
|
||||||
var overflow atomic.Int64
|
|
||||||
var wg sync.WaitGroup
|
|
||||||
wg.Add(1)
|
|
||||||
go func() { defer wg.Done(); deliver(ctx, interval, n, alerts, &overflow) }()
|
|
||||||
// The scan is the only sender: closing alerts tells deliver to drain and
|
|
||||||
// flush once more, and Watch returns only after it has.
|
|
||||||
defer func() { close(alerts); wg.Wait() }()
|
|
||||||
q := alertSender{alerts: alerts, overflow: &overflow}
|
|
||||||
reported := ""
|
reported := ""
|
||||||
for {
|
for {
|
||||||
select {
|
select {
|
||||||
@@ -192,76 +168,7 @@ func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notif
|
|||||||
". Readers may follow symlinks out of the brain; fix them.")
|
". Readers may follow symlinks out of the brain; fix them.")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// alertSender hands alerts to deliver without ever blocking the scan; an
|
|
||||||
// alert that does not fit is counted in overflow, which deliver adds to the
|
|
||||||
// summary.
|
|
||||||
type alertSender struct {
|
|
||||||
alerts chan<- string
|
|
||||||
overflow *atomic.Int64
|
|
||||||
}
|
|
||||||
|
|
||||||
func (s alertSender) push(msg string) {
|
|
||||||
select {
|
|
||||||
case s.alerts <- msg:
|
|
||||||
default:
|
|
||||||
s.overflow.Add(1)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// deliver owns the alertQueue. It flushes as soon as an alert is handed
|
|
||||||
// over, and again every interval while any are pending (outage retries).
|
|
||||||
// When alerts is closed (Watch is stopping) it drains it and flushes once
|
|
||||||
// more, bounded by shutdownFlushTimeout, so an alert handed over just before
|
|
||||||
// shutdown is not lost (PR #180 review). Whatever that flush cannot deliver
|
|
||||||
// is logged as one error with its counts, never dropped silently.
|
|
||||||
func deliver(ctx context.Context, interval time.Duration, n Notifier, alerts <-chan string, overflow *atomic.Int64) {
|
|
||||||
t := time.NewTicker(interval)
|
|
||||||
defer t.Stop()
|
|
||||||
q := &alertQueue{}
|
|
||||||
// drain moves everything handed over into the queue; it reports false
|
|
||||||
// once alerts is closed.
|
|
||||||
drain := func() bool {
|
|
||||||
defer func() { q.suppressed += int(overflow.Swap(0)) }()
|
|
||||||
for {
|
|
||||||
select {
|
|
||||||
case msg, ok := <-alerts:
|
|
||||||
if !ok {
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
q.push(msg)
|
|
||||||
default:
|
|
||||||
return true
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
for {
|
|
||||||
select {
|
|
||||||
case msg, ok := <-alerts:
|
|
||||||
if ok {
|
|
||||||
q.push(msg)
|
|
||||||
ok = drain()
|
|
||||||
}
|
|
||||||
if !ok {
|
|
||||||
q.suppressed += int(overflow.Swap(0))
|
|
||||||
fctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), shutdownFlushTimeout)
|
|
||||||
q.flush(fctx, n)
|
|
||||||
cancel()
|
|
||||||
if len(q.items) > 0 || q.suppressed > 0 {
|
|
||||||
slog.Error("symlink tripwire alerts not delivered at shutdown",
|
|
||||||
"undelivered", len(q.items), "suppressed", q.suppressed)
|
|
||||||
}
|
|
||||||
return
|
|
||||||
}
|
|
||||||
q.flush(ctx, n)
|
q.flush(ctx, n)
|
||||||
case <-t.C:
|
|
||||||
if !drain() {
|
|
||||||
continue // the closed channel is handled by the case above
|
|
||||||
}
|
|
||||||
q.flush(ctx, n)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -1,12 +1,10 @@
|
|||||||
package symlinkguard
|
package symlinkguard
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"bytes"
|
|
||||||
"context"
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"os"
|
"os"
|
||||||
"log/slog"
|
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
@@ -347,19 +345,13 @@ func TestWatchDropsAMessageThatNeverSendsSoLaterAlertsFlow(t *testing.T) {
|
|||||||
}, 3*time.Second, 10*time.Millisecond, "a later alert is delivered despite a poison message")
|
}, 3*time.Second, 10*time.Millisecond, "a later alert is delivered despite a poison message")
|
||||||
}
|
}
|
||||||
|
|
||||||
// toggleNotifier fails while down is set. Like the real audit.NtfyNotifier
|
// toggleNotifier fails while down is set.
|
||||||
// (http.NewRequestWithContext), it also fails, without recording, when its
|
|
||||||
// ctx is already done (PR #180 round 2: notifiers that ignored ctx let a
|
|
||||||
// final flush on the cancelled ctx pass).
|
|
||||||
type toggleNotifier struct {
|
type toggleNotifier struct {
|
||||||
recNotifier
|
recNotifier
|
||||||
down atomic.Bool
|
down atomic.Bool
|
||||||
}
|
}
|
||||||
|
|
||||||
func (r *toggleNotifier) Notify(ctx context.Context, m string) error {
|
func (r *toggleNotifier) Notify(ctx context.Context, m string) error {
|
||||||
if err := ctx.Err(); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
if r.down.Load() {
|
if r.down.Load() {
|
||||||
return errors.New("ntfy down")
|
return errors.New("ntfy down")
|
||||||
}
|
}
|
||||||
@@ -504,271 +496,3 @@ func TestWatchDropsAPoisonMessageAndCountsIt(t *testing.T) {
|
|||||||
return false
|
return false
|
||||||
}, 3*time.Second, 5*time.Millisecond, "the dropped poison alert is counted in a summary")
|
}, 3*time.Second, 5*time.Millisecond, "the dropped poison alert is counted in a summary")
|
||||||
}
|
}
|
||||||
|
|
||||||
// hungNotifier answers every Notify only after latency (or when ctx ends),
|
|
||||||
// with an outage error: an ntfy that accepts connections but never responds
|
|
||||||
// until its timeout.
|
|
||||||
type hungNotifier struct{ latency time.Duration }
|
|
||||||
|
|
||||||
func (h hungNotifier) Notify(ctx context.Context, _ string) error {
|
|
||||||
select {
|
|
||||||
case <-time.After(h.latency):
|
|
||||||
case <-ctx.Done():
|
|
||||||
}
|
|
||||||
return errors.New("ntfy timeout")
|
|
||||||
}
|
|
||||||
|
|
||||||
func waitGone(t *testing.T, link string) {
|
|
||||||
t.Helper()
|
|
||||||
require.Eventually(t, func() bool { _, err := os.Lstat(link); return os.IsNotExist(err) }, 10*time.Second, time.Millisecond)
|
|
||||||
}
|
|
||||||
|
|
||||||
// hyperguild#164: alerts were flushed in the scan goroutine, so with a hung
|
|
||||||
// ntfy and a full queue one flush (6 x notify timeout) delayed the next scan.
|
|
||||||
// The notifier here is slower than the interval, as in production (10 s
|
|
||||||
// timeout vs 30 s interval with 6 messages), so the alert handoff is full:
|
|
||||||
// a scan that waited on it would keep a link for seconds.
|
|
||||||
func TestWatchLinkLifetimeStaysNearOneIntervalWithANotifierSlowerThanTheInterval(t *testing.T) {
|
|
||||||
const interval = 100 * time.Millisecond
|
|
||||||
brain := t.TempDir()
|
|
||||||
ctx, cancel := context.WithCancel(context.Background())
|
|
||||||
defer cancel()
|
|
||||||
go Watch(ctx, brain, interval, hungNotifier{latency: 400 * time.Millisecond})
|
|
||||||
|
|
||||||
for i := 0; i < 2*maxPending+3; i++ { // overfill queue and handoff
|
|
||||||
link := filepath.Join(brain, "fill"+strconv.Itoa(i))
|
|
||||||
require.NoError(t, os.Symlink(t.TempDir(), link))
|
|
||||||
waitGone(t, link)
|
|
||||||
}
|
|
||||||
var worst time.Duration
|
|
||||||
for i := 0; i < 6; i++ {
|
|
||||||
link := filepath.Join(brain, "probe"+strconv.Itoa(i))
|
|
||||||
require.NoError(t, os.Symlink(t.TempDir(), link))
|
|
||||||
planted := time.Now()
|
|
||||||
waitGone(t, link)
|
|
||||||
worst = max(worst, time.Since(planted))
|
|
||||||
}
|
|
||||||
assert.LessOrEqual(t, worst, interval+interval/2, "worst link lifetime %v with a hung notifier", worst)
|
|
||||||
}
|
|
||||||
|
|
||||||
// Review of PR #180: delivery waited for its own tick, about one interval
|
|
||||||
// after the removal. An alert is now flushed as soon as it is handed over.
|
|
||||||
func TestWatchDeliversARemovalAlertWellWithinOneInterval(t *testing.T) {
|
|
||||||
const interval = 400 * time.Millisecond
|
|
||||||
brain := t.TempDir()
|
|
||||||
n := &recNotifier{}
|
|
||||||
ctx, cancel := context.WithCancel(context.Background())
|
|
||||||
defer cancel()
|
|
||||||
go Watch(ctx, brain, interval, n)
|
|
||||||
link := filepath.Join(brain, "prompt")
|
|
||||||
require.NoError(t, os.Symlink(t.TempDir(), link))
|
|
||||||
waitGone(t, link)
|
|
||||||
removed := time.Now()
|
|
||||||
require.Eventually(t, func() bool { return n.count() == 1 }, 2*time.Second, time.Millisecond)
|
|
||||||
assert.Less(t, time.Since(removed), interval/4)
|
|
||||||
}
|
|
||||||
|
|
||||||
// Review of PR #180: on ctx end, deliver returned without draining what the
|
|
||||||
// scan had handed over, so a removal just before shutdown was neither
|
|
||||||
// delivered nor counted. Watch now drains and flushes before returning.
|
|
||||||
func TestWatchDeliversAnAlertHandedOverJustBeforeShutdown(t *testing.T) {
|
|
||||||
for i := 0; i < 20; i++ {
|
|
||||||
brain := t.TempDir()
|
|
||||||
n := &recNotifier{}
|
|
||||||
ctx, cancel := context.WithCancel(context.Background())
|
|
||||||
done := make(chan struct{})
|
|
||||||
go func() { Watch(ctx, brain, 5*time.Millisecond, n); close(done) }()
|
|
||||||
link := filepath.Join(brain, "late")
|
|
||||||
require.NoError(t, os.Symlink(t.TempDir(), link))
|
|
||||||
waitGone(t, link)
|
|
||||||
cancel()
|
|
||||||
<-done
|
|
||||||
require.Equal(t, 1, n.count(), "iteration %d: the removal alert was lost at shutdown", i)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// blockingNotifier holds every Notify until its ctx ends, tracking calls in
|
|
||||||
// flight.
|
|
||||||
type blockingNotifier struct{ inFlight atomic.Int32 }
|
|
||||||
|
|
||||||
func (b *blockingNotifier) Notify(ctx context.Context, _ string) error {
|
|
||||||
b.inFlight.Add(1)
|
|
||||||
defer b.inFlight.Add(-1)
|
|
||||||
<-ctx.Done()
|
|
||||||
return ctx.Err()
|
|
||||||
}
|
|
||||||
|
|
||||||
// Watch does not return while its delivery goroutine is still notifying.
|
|
||||||
func TestWatchReturnsOnlyAfterDeliveryHasStopped(t *testing.T) {
|
|
||||||
prev := shutdownFlushTimeout
|
|
||||||
shutdownFlushTimeout = 50 * time.Millisecond
|
|
||||||
t.Cleanup(func() { shutdownFlushTimeout = prev })
|
|
||||||
brain := t.TempDir()
|
|
||||||
n := &blockingNotifier{}
|
|
||||||
ctx, cancel := context.WithCancel(context.Background())
|
|
||||||
done := make(chan struct{})
|
|
||||||
go func() { Watch(ctx, brain, 5*time.Millisecond, n); close(done) }()
|
|
||||||
link := filepath.Join(brain, "x")
|
|
||||||
require.NoError(t, os.Symlink(t.TempDir(), link))
|
|
||||||
waitGone(t, link)
|
|
||||||
require.Eventually(t, func() bool { return n.inFlight.Load() == 1 }, 2*time.Second, time.Millisecond)
|
|
||||||
cancel()
|
|
||||||
select {
|
|
||||||
case <-done:
|
|
||||||
case <-time.After(5 * time.Second):
|
|
||||||
t.Fatal("Watch did not return after cancel")
|
|
||||||
}
|
|
||||||
assert.Zero(t, n.inFlight.Load(), "Notify still running after Watch returned")
|
|
||||||
}
|
|
||||||
|
|
||||||
// slowToggleNotifier fails slowly (a hung ntfy) while down, then records.
|
|
||||||
type slowToggleNotifier struct {
|
|
||||||
recNotifier
|
|
||||||
down atomic.Bool
|
|
||||||
latency time.Duration
|
|
||||||
}
|
|
||||||
|
|
||||||
func (r *slowToggleNotifier) Notify(ctx context.Context, m string) error {
|
|
||||||
if r.down.Load() {
|
|
||||||
select {
|
|
||||||
case <-time.After(r.latency):
|
|
||||||
case <-ctx.Done():
|
|
||||||
}
|
|
||||||
return errors.New("ntfy timeout")
|
|
||||||
}
|
|
||||||
return r.recNotifier.Notify(ctx, m)
|
|
||||||
}
|
|
||||||
|
|
||||||
// Review of PR #180: with a slow notifier the scan's handoff overflows, and
|
|
||||||
// those alerts exist only in the overflow counter. Every removal is still
|
|
||||||
// delivered or counted in the summary.
|
|
||||||
func TestWatchCountsHandoffOverflowWithASlowNotifier(t *testing.T) {
|
|
||||||
brain := t.TempDir()
|
|
||||||
n := &slowToggleNotifier{latency: 30 * time.Millisecond}
|
|
||||||
n.down.Store(true)
|
|
||||||
ctx, cancel := context.WithCancel(context.Background())
|
|
||||||
defer cancel()
|
|
||||||
go Watch(ctx, brain, 2*time.Millisecond, n)
|
|
||||||
const plants = 40
|
|
||||||
for i := 0; i < plants; i++ {
|
|
||||||
link := filepath.Join(brain, "ov"+strconv.Itoa(i))
|
|
||||||
require.NoError(t, os.Symlink(t.TempDir(), link))
|
|
||||||
waitGone(t, link)
|
|
||||||
}
|
|
||||||
n.down.Store(false)
|
|
||||||
delivered, counted := 0, 0
|
|
||||||
require.Eventually(t, func() bool {
|
|
||||||
n.mu.Lock()
|
|
||||||
defer n.mu.Unlock()
|
|
||||||
delivered, counted = 0, 0
|
|
||||||
for _, m := range n.msgs {
|
|
||||||
if strings.Contains(m, "removed 1 symlink(s)") {
|
|
||||||
delivered++
|
|
||||||
}
|
|
||||||
if i := strings.Index(m, "): "); i >= 0 && strings.Contains(m, "suppressed") {
|
|
||||||
var c int
|
|
||||||
_, _ = fmt.Sscanf(m[i+3:], "%d", &c)
|
|
||||||
counted += c
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return delivered+counted == plants
|
|
||||||
}, 5*time.Second, 10*time.Millisecond, "delivered %d + counted %d of %d", delivered, counted, plants)
|
|
||||||
}
|
|
||||||
|
|
||||||
// An alert still queued when Watch stops (its send failed, the notifier has
|
|
||||||
// since recovered, the retry tick has not come) is flushed once more before
|
|
||||||
// Watch returns, on a ctx that is not the cancelled one: toggleNotifier fails
|
|
||||||
// on a done ctx, as the real ntfy notifier does.
|
|
||||||
func TestWatchFinalFlushSendsAQueuedAlertOnACtxThatOutlivesTheCancel(t *testing.T) {
|
|
||||||
brain := t.TempDir()
|
|
||||||
n := &toggleNotifier{}
|
|
||||||
n.down.Store(true)
|
|
||||||
ctx, cancel := context.WithCancel(context.Background())
|
|
||||||
done := make(chan struct{})
|
|
||||||
go func() { Watch(ctx, brain, 300*time.Millisecond, n); close(done) }()
|
|
||||||
link := filepath.Join(brain, "queued")
|
|
||||||
require.NoError(t, os.Symlink(t.TempDir(), link))
|
|
||||||
waitGone(t, link)
|
|
||||||
time.Sleep(20 * time.Millisecond) // let the immediate send fail
|
|
||||||
n.down.Store(false)
|
|
||||||
cancel()
|
|
||||||
<-done
|
|
||||||
assert.Equal(t, 1, n.count(), "the queued alert was not flushed at shutdown")
|
|
||||||
}
|
|
||||||
|
|
||||||
// lockedBuffer is a goroutine-safe log sink: Watch goroutines of earlier
|
|
||||||
// tests may still log while one test captures.
|
|
||||||
type lockedBuffer struct {
|
|
||||||
mu sync.Mutex
|
|
||||||
buf bytes.Buffer
|
|
||||||
}
|
|
||||||
|
|
||||||
func (b *lockedBuffer) Write(p []byte) (int, error) {
|
|
||||||
b.mu.Lock()
|
|
||||||
defer b.mu.Unlock()
|
|
||||||
return b.buf.Write(p)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (b *lockedBuffer) String() string { b.mu.Lock(); defer b.mu.Unlock(); return b.buf.String() }
|
|
||||||
|
|
||||||
func captureLog(t *testing.T) *lockedBuffer {
|
|
||||||
t.Helper()
|
|
||||||
b := &lockedBuffer{}
|
|
||||||
prev := slog.Default()
|
|
||||||
slog.SetDefault(slog.New(slog.NewTextHandler(b, nil)))
|
|
||||||
t.Cleanup(func() { slog.SetDefault(prev) })
|
|
||||||
return b
|
|
||||||
}
|
|
||||||
|
|
||||||
// PR #180 round 2: when the final flush fails (ntfy hung at shutdown) the
|
|
||||||
// undelivered alerts vanished with only "will retry" warnings. Watch now logs
|
|
||||||
// one error with both counts.
|
|
||||||
func TestWatchLogsUndeliveredAndSuppressedCountsWhenTheFinalFlushFails(t *testing.T) {
|
|
||||||
prev := shutdownFlushTimeout
|
|
||||||
shutdownFlushTimeout = 50 * time.Millisecond
|
|
||||||
t.Cleanup(func() { shutdownFlushTimeout = prev })
|
|
||||||
logs := captureLog(t)
|
|
||||||
brain := t.TempDir()
|
|
||||||
ctx, cancel := context.WithCancel(context.Background())
|
|
||||||
done := make(chan struct{})
|
|
||||||
go func() { Watch(ctx, brain, 5*time.Millisecond, hungNotifier{latency: time.Hour}); close(done) }()
|
|
||||||
link := filepath.Join(brain, "lost")
|
|
||||||
require.NoError(t, os.Symlink(t.TempDir(), link))
|
|
||||||
waitGone(t, link)
|
|
||||||
cancel()
|
|
||||||
<-done
|
|
||||||
out := logs.String()
|
|
||||||
assert.Contains(t, out, "level=ERROR")
|
|
||||||
assert.Contains(t, out, `msg="symlink tripwire alerts not delivered at shutdown"`)
|
|
||||||
assert.Contains(t, out, "undelivered=1")
|
|
||||||
assert.Contains(t, out, "suppressed=0")
|
|
||||||
}
|
|
||||||
|
|
||||||
// deliver, given a handoff that already holds alerts and is closed, moves them
|
|
||||||
// into the queue and runs the final flush, even though the ctx is cancelled:
|
|
||||||
// the drain-on-close path, without depending on when the close lands.
|
|
||||||
func TestDeliverFlushesAlertsDrainedFromAClosedHandoffOnACancelledCtx(t *testing.T) {
|
|
||||||
alerts := make(chan string, 3)
|
|
||||||
alerts <- "a1"
|
|
||||||
alerts <- "a2"
|
|
||||||
close(alerts)
|
|
||||||
ctx, cancel := context.WithCancel(context.Background())
|
|
||||||
cancel()
|
|
||||||
n := &toggleNotifier{}
|
|
||||||
var overflow atomic.Int64
|
|
||||||
deliver(ctx, time.Hour, n, alerts, &overflow)
|
|
||||||
assert.Equal(t, []string{"a1", "a2"}, n.msgs)
|
|
||||||
}
|
|
||||||
|
|
||||||
// PR #180 round 2: when the first receive after close finds the handoff empty,
|
|
||||||
// overflow counted by the scan was never moved into the summary.
|
|
||||||
func TestDeliverCountsOverflowWhenTheHandoffIsClosedAndEmpty(t *testing.T) {
|
|
||||||
alerts := make(chan string)
|
|
||||||
close(alerts)
|
|
||||||
n := &recNotifier{}
|
|
||||||
var overflow atomic.Int64
|
|
||||||
overflow.Store(3)
|
|
||||||
deliver(context.Background(), time.Hour, n, alerts, &overflow)
|
|
||||||
require.Len(t, n.msgs, 1)
|
|
||||||
assert.Contains(t, n.msgs[0], "3 further alert(s)")
|
|
||||||
}
|
|
||||||
|
|||||||
Reference in New Issue
Block a user