test(ingestion): behavioural race tests for raw/ reads; tighten the lint
PR #156 round 3: a readRawNote rewrite to os.ReadFile or Root.ReadFile passed the lint and brought back the escape (19k leaks in 4s) and the FIFO hang. Listing forbidden APIs is whack-a-mole, so the control now rests on behaviour: - TestRawReadersLeakNothingUnderASymlinkSwapRace and TestRawReadersDoNotHangUnderAFIFOSwapRace swap a regular note, a symlink to an outside file, or a FIFO into raw/r.md while ListPending and ReadIngestFile loop for 1.5s; any leaked byte or a hang fails. - TestProcessDir_DoesNotHangUnderAFIFOSwapRace does the same for the unattended watcher (read and archive copy). Flipped: os.ReadFile -> leak + hang tests red; Root.ReadFile -> hang test red; copyFromRoot back to Root.Open -> watcher test red. The lint stays as a second layer: it also forbids ReadFile, matches the selector (so a stored method value is caught), refuses a dot-import of extract, and says it only covers the three files it parses. Refs #153 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,112 @@
|
|||||||
|
package api_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
|
"strings"
|
||||||
|
"sync/atomic"
|
||||||
|
"syscall"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/mathiasbq/hyperguild/ingestion/internal/api"
|
||||||
|
)
|
||||||
|
|
||||||
|
// hyperguild#153 / PR #156: behavioural race tests, so the control does not
|
||||||
|
// rest on lints that list forbidden APIs (round 3: an os.ReadFile or
|
||||||
|
// Root.ReadFile rewrite passed the lint and leaked 19k times in 4s). A
|
||||||
|
// swapper keeps renaming a regular note, a symlink to an outside file, or a
|
||||||
|
// FIFO into raw/r.md while the public readers loop. Any leaked byte fails;
|
||||||
|
// any hang fails.
|
||||||
|
|
||||||
|
const raceSentinel = "SENTINEL-153-RACE"
|
||||||
|
|
||||||
|
func swap(raw, outside string, fifo bool, stop *atomic.Bool) {
|
||||||
|
tgt := filepath.Join(raw, "r.md")
|
||||||
|
for i := 0; !stop.Load(); i++ {
|
||||||
|
tmp := filepath.Join(raw, ".swap")
|
||||||
|
_ = os.Remove(tmp)
|
||||||
|
switch {
|
||||||
|
case i%2 == 0:
|
||||||
|
_ = os.WriteFile(tmp, []byte("# r\n\ninside"), 0o644)
|
||||||
|
case fifo:
|
||||||
|
_ = syscall.Mkfifo(tmp, 0o644)
|
||||||
|
default:
|
||||||
|
_ = os.Symlink(outside, tmp)
|
||||||
|
}
|
||||||
|
_ = os.Rename(tmp, tgt)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func raceBrain(t *testing.T) (brain, raw, outside string) {
|
||||||
|
t.Helper()
|
||||||
|
brain = t.TempDir()
|
||||||
|
raw = filepath.Join(brain, "raw")
|
||||||
|
if err := os.MkdirAll(raw, 0o755); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
outside = filepath.Join(t.TempDir(), "secret.md")
|
||||||
|
if err := os.WriteFile(outside, []byte("# s\n\n"+raceSentinel), 0o644); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if err := os.WriteFile(filepath.Join(raw, "r.md"), []byte("# r\n\ninside"), 0o644); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
return brain, raw, outside
|
||||||
|
}
|
||||||
|
|
||||||
|
// readers runs every public raw/ reader once and returns all text they
|
||||||
|
// produced (excerpts, extracted text).
|
||||||
|
func readers(brain, raw string) string {
|
||||||
|
var b strings.Builder
|
||||||
|
if ns, err := api.ListPending(brain); err == nil {
|
||||||
|
for _, n := range ns {
|
||||||
|
b.WriteString(n.Excerpt)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if s, err := api.ReadIngestFile(brain, filepath.Join(raw, "r.md")); err == nil {
|
||||||
|
b.WriteString(s)
|
||||||
|
}
|
||||||
|
return b.String()
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRawReadersLeakNothingUnderASymlinkSwapRace(t *testing.T) {
|
||||||
|
brain, raw, outside := raceBrain(t)
|
||||||
|
var stop atomic.Bool
|
||||||
|
go swap(raw, outside, false, &stop)
|
||||||
|
defer stop.Store(true)
|
||||||
|
leaks, iters := 0, 0
|
||||||
|
for deadline := time.Now().Add(1500 * time.Millisecond); time.Now().Before(deadline); iters++ {
|
||||||
|
leaks += strings.Count(readers(brain, raw), raceSentinel)
|
||||||
|
}
|
||||||
|
if leaks > 0 {
|
||||||
|
t.Fatalf("outside file leaked %d times in %d iterations", leaks, iters)
|
||||||
|
}
|
||||||
|
t.Logf("%d iterations, 0 leaks", iters)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRawReadersDoNotHangUnderAFIFOSwapRace(t *testing.T) {
|
||||||
|
brain, raw, outside := raceBrain(t)
|
||||||
|
var stop atomic.Bool
|
||||||
|
go swap(raw, outside, true, &stop)
|
||||||
|
var iters atomic.Int64
|
||||||
|
done := make(chan struct{})
|
||||||
|
go func() {
|
||||||
|
for deadline := time.Now().Add(1500 * time.Millisecond); time.Now().Before(deadline); {
|
||||||
|
_ = readers(brain, raw)
|
||||||
|
iters.Add(1)
|
||||||
|
}
|
||||||
|
close(done)
|
||||||
|
}()
|
||||||
|
select {
|
||||||
|
case <-done:
|
||||||
|
stop.Store(true)
|
||||||
|
case <-time.After(6 * time.Second):
|
||||||
|
stop.Store(true)
|
||||||
|
// Unblock the stuck open so the test process can exit.
|
||||||
|
if f, err := os.OpenFile(filepath.Join(raw, "r.md"), os.O_WRONLY|syscall.O_NONBLOCK, 0); err == nil {
|
||||||
|
_ = f.Close()
|
||||||
|
}
|
||||||
|
t.Fatalf("a raw/ reader hung on a swapped-in FIFO after %d iterations", iters.Load())
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -58,6 +58,9 @@ func TestNoPackageReferencesExtractTextOutsideExtract(t *testing.T) {
|
|||||||
if imp.Name != nil {
|
if imp.Name != nil {
|
||||||
name = imp.Name.Name
|
name = imp.Name.Name
|
||||||
}
|
}
|
||||||
|
if name == "." {
|
||||||
|
t.Errorf("%s dot-imports extract; import it by name so its use stays checkable", path)
|
||||||
|
}
|
||||||
names[name] = true
|
names[name] = true
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -77,13 +80,21 @@ func TestNoPackageReferencesExtractTextOutsideExtract(t *testing.T) {
|
|||||||
|
|
||||||
// PR #156 review round 2: the FIFO fix (open non-blocking, check the handle)
|
// PR #156 review round 2: the FIFO fix (open non-blocking, check the handle)
|
||||||
// lives in OpenRegular, and reverting any caller to a plain Root.Open stayed
|
// lives in OpenRegular, and reverting any caller to a plain Root.Open stayed
|
||||||
// green. In the raw/ readers, the only Open/OpenFile on anything is inside
|
// green. In the three raw/ reader FILES parsed below (and only those: a
|
||||||
// OpenRegular itself.
|
// wrapper in another file is not seen), no Open/OpenFile/ReadFile selector
|
||||||
|
// may appear outside OpenRegular, called or taken as a method value. This is
|
||||||
|
// a second layer; the behavioural race tests in internal/api and
|
||||||
|
// internal/watcher are what catch a rewrite through any API (round 3).
|
||||||
// osOpenAllowed names the functions in guarded files that may call os.Open*
|
// osOpenAllowed names the functions in guarded files that may call os.Open*
|
||||||
// on a path that is not a raw/ read.
|
// on a path that is not a raw/ read.
|
||||||
var osOpenAllowed = map[string]bool{"internal/watcher/watcher.go:appendWatcherLog": true}
|
var osOpenAllowed = map[string]bool{
|
||||||
|
"internal/watcher/watcher.go:appendWatcherLog": true, // ingest log append, not a raw/ read
|
||||||
|
// Text reads by path; outside this package only TextIn's temp copy reaches
|
||||||
|
// it (TestNoPackageReferencesExtractTextOutsideExtract enforces that).
|
||||||
|
"internal/extract/extract.go:Text": true,
|
||||||
|
}
|
||||||
|
|
||||||
func TestRawReadersOpenOnlyThroughOpenRegular(t *testing.T) {
|
func TestRawReaderFilesUseNoOpenOrReadFileOutsideOpenRegular(t *testing.T) {
|
||||||
guarded := []string{"internal/api/rawroot.go", "internal/extract/extract.go", "internal/watcher/watcher.go"}
|
guarded := []string{"internal/api/rawroot.go", "internal/extract/extract.go", "internal/watcher/watcher.go"}
|
||||||
files := parseNonTest(t)
|
files := parseNonTest(t)
|
||||||
for _, g := range guarded {
|
for _, g := range guarded {
|
||||||
@@ -102,11 +113,11 @@ func TestRawReadersOpenOnlyThroughOpenRegular(t *testing.T) {
|
|||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
ast.Inspect(fn, func(n ast.Node) bool {
|
ast.Inspect(fn, func(n ast.Node) bool {
|
||||||
call, ok := n.(*ast.CallExpr)
|
sel, ok := n.(*ast.SelectorExpr)
|
||||||
if !ok {
|
if !ok {
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
if sel, ok := call.Fun.(*ast.SelectorExpr); ok && (sel.Sel.Name == "Open" || sel.Sel.Name == "OpenFile") {
|
if sel.Sel.Name == "Open" || sel.Sel.Name == "OpenFile" || sel.Sel.Name == "ReadFile" {
|
||||||
if id, ok := sel.X.(*ast.Ident); ok && id.Name == "os" && osOpenAllowed[g+":"+fn.Name.Name] {
|
if id, ok := sel.X.(*ast.Ident); ok && id.Name == "os" && osOpenAllowed[g+":"+fn.Name.Name] {
|
||||||
return true // a named non-raw write (the ingest log append)
|
return true // a named non-raw write (the ingest log append)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -8,6 +8,8 @@ import (
|
|||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"strings"
|
"strings"
|
||||||
|
"sync/atomic"
|
||||||
|
"syscall"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -334,3 +336,51 @@ func TestProcessDir_RefusesASymlinkedRawDir(t *testing.T) {
|
|||||||
assert.NotEmpty(t, errs, "a symlinked raw/ must be reported, not silently ingested")
|
assert.NotEmpty(t, errs, "a symlinked raw/ must be reported, not silently ingested")
|
||||||
assert.Zero(t, calls)
|
assert.Zero(t, calls)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// PR #156: a FIFO swapped into raw/ between the watcher's checks and its
|
||||||
|
// read or archive copy must not hang the watcher, which runs unattended.
|
||||||
|
func TestProcessDir_DoesNotHangUnderAFIFOSwapRace(t *testing.T) {
|
||||||
|
brainDir := setupBrainDir(t)
|
||||||
|
raw := filepath.Join(brainDir, "raw")
|
||||||
|
var stop atomic.Bool
|
||||||
|
go func() {
|
||||||
|
for i := 0; !stop.Load(); i++ {
|
||||||
|
tmp := filepath.Join(raw, ".swap")
|
||||||
|
_ = os.Remove(tmp)
|
||||||
|
if i%2 == 0 {
|
||||||
|
_ = os.WriteFile(tmp, []byte("x"), 0o644)
|
||||||
|
} else {
|
||||||
|
_ = syscall.Mkfifo(tmp, 0o644)
|
||||||
|
}
|
||||||
|
_ = os.Rename(tmp, filepath.Join(raw, "r.md"))
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
cfg := Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: pipeline.Config{
|
||||||
|
Complete: func(context.Context, string, string) (string, error) {
|
||||||
|
b, _ := json.Marshal([]pipeline.RawPage{{Title: "R", Type: "source", Subtype: "article", Content: "## S\n\nx\n"}})
|
||||||
|
return string(b), nil
|
||||||
|
},
|
||||||
|
Schema: "# Schema",
|
||||||
|
}}
|
||||||
|
var iters atomic.Int64
|
||||||
|
done := make(chan struct{})
|
||||||
|
go func() {
|
||||||
|
for deadline := time.Now().Add(1500 * time.Millisecond); time.Now().Before(deadline); {
|
||||||
|
_ = os.Remove(filepath.Join(raw, "r.md.processed"))
|
||||||
|
_ = os.Remove(filepath.Join(raw, "r.md.failed"))
|
||||||
|
_ = processDir(context.Background(), cfg, "2026-10-03")
|
||||||
|
iters.Add(1)
|
||||||
|
}
|
||||||
|
close(done)
|
||||||
|
}()
|
||||||
|
select {
|
||||||
|
case <-done:
|
||||||
|
stop.Store(true)
|
||||||
|
case <-time.After(8 * time.Second):
|
||||||
|
stop.Store(true)
|
||||||
|
if f, err := os.OpenFile(filepath.Join(raw, "r.md"), os.O_WRONLY|syscall.O_NONBLOCK, 0); err == nil {
|
||||||
|
_ = f.Close()
|
||||||
|
}
|
||||||
|
t.Fatalf("the watcher hung on a swapped-in FIFO after %d iterations", iters.Load())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user