Jakub Zych
fda46466c7
A reaper pass that cannot read the store keeps the copy by design, so the Redis case reaps until the copy is gone and still requires exactly one close per instance; both copies are asserted before the expiry.
262 lines
8.8 KiB
Go
262 lines
8.8 KiB
Go
package pier_test
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"io"
|
|
"log/slog"
|
|
"net/http"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.golem15.com/golem15/summercms/modules/pact"
|
|
)
|
|
|
|
// logBuffer is a goroutine-safe slog destination for audit-line checks.
|
|
type logBuffer struct {
|
|
mu sync.Mutex
|
|
buf bytes.Buffer
|
|
}
|
|
|
|
func (b *logBuffer) Write(p []byte) (int, error) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
return b.buf.Write(p)
|
|
}
|
|
|
|
func (b *logBuffer) String() string {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
return b.buf.String()
|
|
}
|
|
|
|
// captureLogs routes the process logger (and so diary's pier component)
|
|
// into a buffer for the rest of the test.
|
|
func captureLogs(t *testing.T) *logBuffer {
|
|
t.Helper()
|
|
buf := &logBuffer{}
|
|
previous := slog.Default()
|
|
slog.SetDefault(slog.New(slog.NewTextHandler(buf, &slog.HandlerOptions{Level: slog.LevelDebug})))
|
|
t.Cleanup(func() { slog.SetDefault(previous) })
|
|
return buf
|
|
}
|
|
|
|
// TestSessionReaper pins the session reaper of plan 06 (A-CR-01, A-IN-01,
|
|
// A-IN-06): a stateful session whose store record expired is closed on the
|
|
// instance holding it (its SDK session ends, the id answers 404, the audit
|
|
// line says expired); on a shared Redis store an instance keeps a copy only
|
|
// while the record lives; a store error never closes a copy; a GET stream
|
|
// whose record vanished ends through the re-check; and a tool call served
|
|
// without a session logs the session as ephemeral.
|
|
func TestSessionReaper(t *testing.T) {
|
|
plugin := acmePlugin{tools: []pact.MCPTool{echoTool()}}
|
|
|
|
t.Run("memory store reaps an idle session past the TTL", func(t *testing.T) {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
|
defer cancel()
|
|
in := startInstance(t, newApp(t, appSpec{
|
|
mcp: "enabled: false\nsessions:\n store: memory\n idle_ttl: 30m\n",
|
|
guards: []pact.MCPGuard{&acmeGuard{}},
|
|
}), plugin)
|
|
p := in.pier()
|
|
t.Cleanup(p.StopReaper)
|
|
logs := captureLogs(t)
|
|
|
|
sid := openSession(t, ctx, in, "oas_at_test")
|
|
done, ok := p.SessionDone(sid)
|
|
if !ok {
|
|
t.Fatal("the session is not connected on the instance that opened it")
|
|
}
|
|
if n := p.ReapNow(ctx); n != 0 {
|
|
t.Fatalf("a pass within the TTL reaped %d sessions", n)
|
|
}
|
|
if !p.AdvanceStoreClock(31 * time.Minute) {
|
|
t.Fatal("the instance has no memory store")
|
|
}
|
|
if n := p.ReapNow(ctx); n != 1 {
|
|
t.Fatalf("reaped = %d, want 1", n)
|
|
}
|
|
select {
|
|
case <-done:
|
|
case <-time.After(5 * time.Second):
|
|
t.Fatal("ss.Wait did not return after the reap")
|
|
}
|
|
if n := p.LocalSessions(); n != 0 {
|
|
t.Fatalf("local sessions after the reap = %d, want 0", n)
|
|
}
|
|
if gone := rpc(t, ctx, in.url(), "oas_at_test", sid, callBody); gone.status != http.StatusNotFound {
|
|
t.Fatalf("call on the reaped session = %d %s", gone.status, gone.body)
|
|
}
|
|
logged := logs.String()
|
|
if !strings.Contains(logged, "step=expired") || !strings.Contains(logged, "session="+sid[:8]) {
|
|
t.Fatalf("no expired audit line for %s: %s", sid[:8], logged)
|
|
}
|
|
if strings.Contains(logged, "oas_at_test") {
|
|
t.Fatalf("the audit lines carry the token: %s", logged)
|
|
}
|
|
})
|
|
|
|
t.Run("redis instance keeps its copy only while the record lives", func(t *testing.T) {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
|
defer cancel()
|
|
a, b, mr := twoInstances(t, appSpec{
|
|
mcp: "enabled: false\nsessions:\n store: redis\n idle_ttl: 30m\n",
|
|
guards: []pact.MCPGuard{&acmeGuard{}},
|
|
}, plugin)
|
|
t.Cleanup(a.pier().StopReaper)
|
|
t.Cleanup(b.pier().StopReaper)
|
|
|
|
sid := openSession(t, ctx, a, "oas_at_test")
|
|
// B serves one request and holds a rehydrated copy; A keeps serving.
|
|
if call := rpc(t, ctx, b.url(), "oas_at_test", sid, callBody); call.status != http.StatusOK {
|
|
t.Fatalf("call on B = %d %s", call.status, call.body)
|
|
}
|
|
if n := b.pier().LocalSessions(); n != 1 {
|
|
t.Fatalf("B local sessions = %d, want 1", n)
|
|
}
|
|
mr.FastForward(20 * time.Minute)
|
|
if call := rpc(t, ctx, a.url(), "oas_at_test", sid, callBody); call.status != http.StatusOK {
|
|
t.Fatalf("call on A = %d %s", call.status, call.body)
|
|
}
|
|
mr.FastForward(20 * time.Minute)
|
|
if n := b.pier().ReapNow(ctx); n != 0 {
|
|
t.Fatalf("B reaped a session A kept alive (%d closed)", n)
|
|
}
|
|
if na, nb := a.pier().LocalSessions(), b.pier().LocalSessions(); na != 1 || nb != 1 {
|
|
t.Fatalf("local sessions while the record lives: A %d, B %d, want 1 and 1", na, nb)
|
|
}
|
|
mr.FastForward(31 * time.Minute)
|
|
// A pass that cannot read the store keeps the copy (asserted below),
|
|
// so a loaded host may need a second pass; the copy is reaped once.
|
|
reapAll := func(name string, in *instance) {
|
|
t.Helper()
|
|
closed := 0
|
|
deadline := time.Now().Add(5 * time.Second)
|
|
for in.pier().LocalSessions() != 0 {
|
|
if time.Now().After(deadline) {
|
|
t.Fatalf("%s still holds %d copies after the record expired", name, in.pier().LocalSessions())
|
|
}
|
|
closed += in.pier().ReapNow(ctx)
|
|
}
|
|
if closed != 1 {
|
|
t.Fatalf("%s reaped %d sessions after the record expired, want 1", name, closed)
|
|
}
|
|
}
|
|
reapAll("B", b)
|
|
reapAll("A", a)
|
|
if a.pier().LocalSessions() != 0 || b.pier().LocalSessions() != 0 {
|
|
t.Fatalf("copies left: A %d, B %d", a.pier().LocalSessions(), b.pier().LocalSessions())
|
|
}
|
|
})
|
|
|
|
t.Run("a store error keeps the copy", func(t *testing.T) {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
|
defer cancel()
|
|
a, _, mr := twoInstances(t, appSpec{
|
|
mcp: "enabled: false\nsessions:\n store: redis\n idle_ttl: 30m\n",
|
|
guards: []pact.MCPGuard{&acmeGuard{}},
|
|
}, plugin)
|
|
t.Cleanup(a.pier().StopReaper)
|
|
logs := captureLogs(t)
|
|
openSession(t, ctx, a, "oas_at_test")
|
|
openSession(t, ctx, a, "oas_at_test")
|
|
mr.SetError("ERR the store is down")
|
|
n := a.pier().ReapNow(ctx)
|
|
mr.SetError("")
|
|
if n != 0 {
|
|
t.Fatalf("a store error closed %d sessions", n)
|
|
}
|
|
if got := a.pier().LocalSessions(); got != 2 {
|
|
t.Fatalf("local sessions after a failing pass = %d, want 2", got)
|
|
}
|
|
if c := strings.Count(logs.String(), "session reaper could not read the store"); c != 1 {
|
|
t.Fatalf("store error logged %d times in one pass, want once: %s", c, logs.String())
|
|
}
|
|
})
|
|
|
|
t.Run("a GET stream whose record vanished ends through the re-check", func(t *testing.T) {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
|
defer cancel()
|
|
guard := &acmeGuard{}
|
|
in := startInstance(t, newApp(t, appSpec{
|
|
mcp: "enabled: false\nsessions:\n store: memory\n idle_ttl: 30m\n",
|
|
guards: []pact.MCPGuard{guard},
|
|
}), plugin)
|
|
p := in.pier()
|
|
t.Cleanup(p.StopReaper)
|
|
p.SetRecheck(50 * time.Millisecond)
|
|
logs := captureLogs(t)
|
|
|
|
sid := openSession(t, ctx, in, "oas_at_test")
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodGet, in.url(), nil)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
req.Header.Set("Authorization", "Bearer oas_at_test")
|
|
req.Header.Set("Accept", "text/event-stream")
|
|
req.Header.Set("Mcp-Session-Id", sid)
|
|
req.Header.Set("Mcp-Protocol-Version", "2025-11-25")
|
|
res, err := http.DefaultClient.Do(req)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer res.Body.Close()
|
|
if res.StatusCode != http.StatusOK {
|
|
t.Fatalf("GET stream = %d", res.StatusCode)
|
|
}
|
|
ended := make(chan []byte, 1)
|
|
go func() {
|
|
raw, _ := io.ReadAll(res.Body)
|
|
ended <- raw
|
|
}()
|
|
select {
|
|
case <-ended:
|
|
t.Fatal("the stream ended while its record lived")
|
|
case <-time.After(300 * time.Millisecond):
|
|
}
|
|
// The record expires; no event is published, so only the re-check
|
|
// can notice.
|
|
if !p.AdvanceStoreClock(31 * time.Minute) {
|
|
t.Fatal("the instance has no memory store")
|
|
}
|
|
select {
|
|
case raw := <-ended:
|
|
if strings.Contains(string(raw), "ended") {
|
|
t.Fatalf("the stream delivered an event before closing: %s", raw)
|
|
}
|
|
case <-time.After(5 * time.Second):
|
|
t.Fatal("the stream outlived its record")
|
|
}
|
|
if n := p.LocalSessions(); n != 0 {
|
|
t.Fatalf("local sessions after the stream closed = %d, want 0", n)
|
|
}
|
|
if !strings.Contains(logs.String(), "step=stream_closed") {
|
|
t.Fatalf("no stream_closed audit line: %s", logs.String())
|
|
}
|
|
})
|
|
|
|
t.Run("a call without a session logs ephemeral", func(t *testing.T) {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
|
defer cancel()
|
|
in := startInstance(t, newApp(t, appSpec{
|
|
mcp: "enabled: false\nsessions:\n store: memory\n",
|
|
guards: []pact.MCPGuard{&acmeGuard{}},
|
|
}), plugin)
|
|
t.Cleanup(in.pier().StopReaper)
|
|
logs := captureLogs(t)
|
|
call := rpc(t, ctx, in.url(), "oas_at_test", "", callBody)
|
|
if call.status != http.StatusOK || !strings.Contains(call.body, `\"principal\":7`) {
|
|
t.Fatalf("call without a session = %d %s", call.status, call.body)
|
|
}
|
|
logged := logs.String()
|
|
if !strings.Contains(logged, "pier: tool call") || !strings.Contains(logged, "session=ephemeral") {
|
|
t.Fatalf("the tool call is not logged as ephemeral: %s", logged)
|
|
}
|
|
if n := in.pier().LocalSessions(); n != 0 {
|
|
t.Fatalf("an ephemeral call left %d local sessions", n)
|
|
}
|
|
})
|
|
}
|