hyperguild/ingestion/internal/api/raw_race_test.go

package api_test

import (
	"encoding/base64"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"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"

// startSwapper runs fn until the test ends, and waits for it to exit before
// t.TempDir's cleanup removes the directory it writes into (hyperguild#162:
// an unawaited swapper raced the cleanup, "directory not empty").
func startSwapper(t *testing.T, fn func(stop *atomic.Bool)) {
	t.Helper()
	var stop atomic.Bool
	var wg sync.WaitGroup
	wg.Add(1)
	go func() { defer wg.Done(); fn(&stop) }()
	t.Cleanup(func() { stop.Store(true); wg.Wait() })
}

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)
	startSwapper(t, func(stop *atomic.Bool) { swap(raw, outside, false, stop) })
	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)
	startSwapper(t, func(stop *atomic.Bool) { 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:
	case <-time.After(6 * time.Second):
		// 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())
	}
}

// PR #156 round 4: the pdf/docmark branch of TextIn copies the file through
// the root to a temp file; pointing the extractor back at the raw/ path by
// name stayed green because the race tests only used r.md (a stub docmark
// then received the outside file's bytes 34k times in 4s). This races a
// symlink into raw/r.docx and has a fake docmark count the sentinel in every
// document it is sent.
func TestDocmarkExtractionLeaksNothingUnderASymlinkSwapRace(t *testing.T) {
	var seen atomic.Int64
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		var req struct {
			Params struct {
				Arguments struct {
					ContentBase64 string `json:"content_base64"`
				} `json:"arguments"`
			} `json:"params"`
		}
		_ = json.NewDecoder(r.Body).Decode(&req)
		if b, err := base64.StdEncoding.DecodeString(req.Params.Arguments.ContentBase64); err == nil &&
			strings.Contains(string(b), raceSentinel) {
			seen.Add(1)
		}
		w.Header().Set("Content-Type", "application/json")
		_, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":1,"result":{"content":[{"type":"text","text":"converted"}]}}`))
	}))
	defer srv.Close()
	t.Setenv("DOCMARK_URL", srv.URL)

	brain, raw, outside := raceBrain(t)
	tgt := filepath.Join(raw, "r.docx")
	if err := os.WriteFile(tgt, []byte("inside docx bytes"), 0o644); err != nil {
		t.Fatal(err)
	}
	startSwapper(t, func(stop *atomic.Bool) {
		for i := 0; !stop.Load(); i++ {
			tmp := filepath.Join(raw, ".swapdocx")
			_ = os.Remove(tmp)
			if i%2 == 0 {
				_ = os.WriteFile(tmp, []byte("inside docx bytes"), 0o644)
			} else {
				_ = os.Symlink(outside, tmp)
			}
			_ = os.Rename(tmp, tgt)
		}
	})
	iters := 0
	for deadline := time.Now().Add(1500 * time.Millisecond); time.Now().Before(deadline); iters++ {
		_, _ = api.ReadIngestFile(brain, tgt)
	}
	if n := seen.Load(); n > 0 {
		t.Fatalf("docmark received the outside file %d times in %d iterations", n, iters)
	}
	t.Logf("%d iterations, 0 leaks to docmark", iters)
}