Files
Maven/internal/ipc/ipc_test.go
T
kami f1a809121b shutdown: close the sockets, or the database never gets sealed
mavend seals its encrypted database in `defer st.Close()` when run() returns.
It had not returned since 2026-07-21. Every restart since then decrypted the
same eleven-day-old ciphertext and rolled back everything written in between:
the Telegram nudge that kept firing was a fact being un-written on each boot.

The goroutine dump named it. main → srv.Close() → ipc.(*Server).Close →
wg.Wait(), waiting on per-connection goroutines parked in readFrame. Close
shut the listener and nothing else, so the idle persistent sockets held by
mavweb, mavpoll, mavcaldav and mavmaild blocked shutdown forever. `docker
compose stop -t 60` spent the whole sixty seconds and then took a SIGKILL.

So: track the accepted conns and close them, in ipc and in voice, which had
the identical defect. Bound all three waits — the two per-server ones and the
worker wait in main — because the seal matters more than any single in-flight
call. A dropped RPC costs one reply; a missed seal costs a session.

The regression test leaves a client connected and idle, which is the case the
old tests avoided by closing the client first.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TrVSBKe3RFDF4fGYKWYQnX
2026-08-01 20:05:52 +04:00

670 lines
22 KiB
Go

package ipc
import (
"bytes"
"context"
"encoding/binary"
"encoding/json"
"errors"
"fmt"
"io"
"net"
"os"
"path/filepath"
"testing"
"time"
"github.com/kami/maven/internal/store"
)
// tmpSocket — a socket path under a 0700 temp dir, unique per test.
func tmpSocket(t *testing.T) string {
t.Helper()
dir := t.TempDir()
return filepath.Join(dir, "maven.sock")
}
// newServerWithStore spins a real store + Server + Client so the boundary
// is exercised exactly as the daemon wires it. Returns the api (for direct
// in-process expectations) and a client going through the socket.
func newServerWithStore(t *testing.T) (CoreAPI, *Server, *Client, *store.Store) {
t.Helper()
dir := t.TempDir()
s, err := store.Open(context.Background(), filepath.Join(dir, "maven.db"))
if err != nil {
t.Fatalf("open store: %v", err)
}
t.Cleanup(func() { _ = s.Close() })
api := NewStoreAPI(s)
srv, err := Listen(tmpSocket(t), api)
if err != nil {
t.Fatalf("listen: %v", err)
}
done := make(chan struct{})
go func() {
_ = srv.Serve()
close(done)
}()
t.Cleanup(func() {
_ = srv.Close()
<-done
})
cli, err := Dial(srv.Path())
if err != nil {
t.Fatalf("dial: %v", err)
}
t.Cleanup(func() { _ = cli.Close() })
return api, srv, cli, s
}
// TestDialWait_WaitsForLateServer — a module may start before core's socket
// exists (core loads models first). DialWait must keep retrying until the
// socket appears rather than fail on the first attempt.
func TestDialWait_WaitsForLateServer(t *testing.T) {
dir := t.TempDir()
sock := filepath.Join(dir, "maven.sock")
srvCh := make(chan *Server, 1)
errCh := make(chan error, 1)
// bring the server up only after DialWait is already retrying
go func() {
time.Sleep(300 * time.Millisecond)
s, err := store.Open(context.Background(), filepath.Join(dir, "maven.db"))
if err != nil {
errCh <- err
return
}
t.Cleanup(func() { _ = s.Close() })
srv, err := Listen(sock, NewStoreAPI(s))
if err != nil {
errCh <- err
return
}
go func() { _ = srv.Serve() }()
srvCh <- srv
}()
cli, err := DialWait(sock, 5*time.Second)
if err != nil {
t.Fatalf("DialWait should connect once the server appears: %v", err)
}
select {
case err := <-errCh:
t.Fatalf("server setup failed: %v", err)
case srv := <-srvCh:
_ = cli.Close() // close client first so srv.Close's handler wait returns
_ = srv.Close()
}
}
// TestClient_ReconnectsAfterServerRestart — a long-lived module (e.g. mavweb)
// must survive a core restart. The first server is closed and a new one is
// brought up on the SAME socket path (as a daemon restart does); the client's
// cached conn is now dead. The next call must transparently re-dial and succeed
// instead of failing forever with "broken pipe".
func TestClient_ReconnectsAfterServerRestart(t *testing.T) {
dir := t.TempDir()
sock := filepath.Join(dir, "maven.sock")
serve := func() *Server {
s, err := store.Open(context.Background(), filepath.Join(dir, "maven.db"))
if err != nil {
t.Fatalf("open store: %v", err)
}
t.Cleanup(func() { _ = s.Close() })
srv, err := Listen(sock, NewStoreAPI(s))
if err != nil {
t.Fatalf("listen: %v", err)
}
go func() { _ = srv.Serve() }()
return srv
}
srv1 := serve()
cli, err := Dial(sock)
if err != nil {
t.Fatalf("dial: %v", err)
}
t.Cleanup(func() { _ = cli.Close() })
// works against the first server
if _, err := cli.Presence(context.Background()); err != nil {
t.Fatalf("call before restart: %v", err)
}
// Simulate a core restart: the client's conn dies (as it would when the
// daemon process exits), then a fresh server binds the SAME path. Closing
// the client side first also lets srv1's handler goroutine see EOF and
// exit, so srv1.Close()'s wg.Wait() returns instead of blocking on a
// parked reader.
cli.conn.Close()
_ = srv1.Close()
srv2 := serve()
// The cached conn is dead — the call must transparently re-dial and succeed.
if _, err := cli.Presence(context.Background()); err != nil {
t.Fatalf("call after restart should have re-dialed, got: %v", err)
}
// Teardown order matters: Server.Close waits for its handler goroutine,
// which is parked reading the (now live, re-dialed) client conn. Close the
// client first so the handler sees EOF and Close returns instead of hanging.
_ = cli.Close()
_ = srv2.Close()
}
// TestFrame_Roundtrip — JSON over a length prefix survives the loop, and the
// prefix itself encodes the length exactly. The framing is the only thing
// keeping a module's request paired with core's reply; it's worth a direct test.
func TestFrame_Roundtrip(t *testing.T) {
var buf bytes.Buffer
type payload struct {
Msg string `json:"m"`
N int `json:"n"`
}
want := payload{Msg: "hello", N: 42}
if err := writeFrame(&buf, want); err != nil {
t.Fatalf("writeFrame: %v", err)
}
// header length must equal the JSON body length that follows.
var hdr [4]byte
if _, err := io.ReadFull(&buf, hdr[:]); err != nil {
t.Fatalf("read hdr: %v", err)
}
bodyLen := binary.BigEndian.Uint32(hdr[:])
if int(bodyLen) != buf.Len() {
t.Fatalf("prefix length %d != body %d", bodyLen, buf.Len())
}
// readFrame consumes header+body together; recombine so it sees a whole frame.
full := append(hdr[:], buf.Bytes()...)
var got payload
if err := readFrame(bytes.NewReader(full), &got); err != nil {
t.Fatalf("readFrame: %v", err)
}
if got != want {
t.Fatalf("roundtrip mismatch: got %+v want %+v", got, want)
}
}
func TestFrame_TooLarge(t *testing.T) {
// Encode-side guard refuses to ship anything bigger than maxFrame; the
// socket never sees it. Defense against a confused peer, not a real path.
big := make([]byte, maxFrame+1)
if err := writeFrame(io.Discard, big); !errors.Is(err, ErrFrameTooLarge) {
t.Fatalf("writeFrame: got %v, want ErrFrameTooLarge", err)
}
// Decode-side guard refuses a header claiming a too-large body; the conn
// is now desynced (length read, body not), but readFrame doesn't have to
// recover — the caller closes it.
var hdr [4]byte
binary.BigEndian.PutUint32(hdr[:], maxFrame+1)
if err := readFrame(bytes.NewReader(hdr[:]), nil); !errors.Is(err, ErrFrameTooLarge) {
t.Fatalf("readFrame: got %v, want ErrFrameTooLarge", err)
}
}
// TestSocket_Perms — the auth floor. 0600 ⇒ only the same unix user can
// connect. If this regresses to world-readable, every user on the box is a
// module; that's the entire auth model today, so assert it.
func TestSocket_Perms(t *testing.T) {
_, srv, _, _ := newServerWithStore(t)
fi, err := os.Stat(srv.Path())
if err != nil {
t.Fatalf("stat socket: %v", err)
}
mode := fi.Mode().Perm()
if mode != 0o600 {
t.Fatalf("socket perm = %#o, want 0600", mode)
}
}
// TestStoreAPI_Direct — the in-process adapter path (no socket) maps store
// sentinels to ipc sentinels. The boundary's contract is that error identity
// is the same on both sides; this pins it for the daemon-embedded modules
// (router, delivery today) that never go over the wire.
func TestStoreAPI_Direct(t *testing.T) {
api, _, _, _ := newServerWithStore(t)
ctx := context.Background()
// missing key ⇒ ErrNoFact
if _, err := api.LatestFact(ctx, "nope"); !errors.Is(err, ErrNoFact) {
t.Fatalf("LatestFact missing: got %v, want ErrNoFact", err)
}
// bad confidence ⇒ ErrConfidence
if _, err := api.WriteFact(ctx, WriteFactReq{
Ts: time.Now(), Kind: "self", Key: "water", Value: "1", Source: "tap:water", Confidence: 0,
}); !errors.Is(err, ErrConfidence) {
t.Fatalf("WriteFact conf=0: got %v, want ErrConfidence", err)
}
// since on missing key ⇒ ErrNoFact
if _, err := api.Since(ctx, "nope", time.Now()); !errors.Is(err, ErrNoFact) {
t.Fatalf("Since missing: got %v, want ErrNoFact", err)
}
// reminder idempotency: invalid status ⇒ ErrReminderState
if err := api.MarkReminder(ctx, 99999, "weird"); !errors.Is(err, ErrReminderState) {
t.Fatalf("MarkReminder weird: got %v, want ErrReminderState", err)
}
// resolve nonexistent nudge ⇒ ErrNudgeNotFound
if err := api.ResolveNudge(ctx, 99999, "acted", time.Now()); !errors.Is(err, ErrNudgeNotFound) {
t.Fatalf("ResolveNudge none: got %v, want ErrNudgeNotFound", err)
}
}
// TestClient_E2E — full socket round trip against a real store. Drives every
// method end-to-end and asserts sentinel identity survives the wire. This is
// the test that catches the boundary bugs: param shape mismatch, sentinel
// code drift, dto mapping, framing interleaving.
func TestClient_E2E(t *testing.T) {
_, _, cli, _ := newServerWithStore(t)
ctx := context.Background()
now := time.Now().UTC().Truncate(time.Millisecond)
// write a tap (self, confidence 1.0) and read it back.
id, err := cli.WriteFact(ctx, WriteFactReq{
Ts: now, Kind: "self", Key: "water", Value: "1", Source: "tap:water", Confidence: 1.0,
})
if err != nil {
t.Fatalf("WriteFact: %v", err)
}
if id <= 0 {
t.Fatalf("WriteFact returned id %d", id)
}
f, err := cli.LatestFact(ctx, "water")
if err != nil {
t.Fatalf("LatestFact: %v", err)
}
if f.Key != "water" || f.Value != "1" || f.Source != "tap:water" || f.Confidence != 1.0 {
t.Fatalf("LatestFact mismatch: %+v", f)
}
if !f.Ts.Equal(now) {
t.Fatalf("Ts roundtrip: got %v want %v", f.Ts, now)
}
// provenance scope: a foreign source doesn't see the tap value.
if _, err := cli.LatestFactBySource(ctx, "water", "poll:evil"); !errors.Is(err, ErrNoFact) {
t.Fatalf("LatestFactBySource foreign: got %v, want ErrNoFact", err)
}
if _, err := cli.LatestFactBySource(ctx, "water", "tap:water"); err != nil {
t.Fatalf("LatestFactBySource own: %v", err)
}
// since: ~0 elapsed since "now".
d, err := cli.Since(ctx, "water", now.Add(time.Second))
if err != nil {
t.Fatalf("Since: %v", err)
}
if d != time.Second {
t.Fatalf("Since dur = %v, want 1s", d)
}
// since missing ⇒ ErrNoFact over the wire.
if _, err := cli.Since(ctx, "nope", now); !errors.Is(err, ErrNoFact) {
t.Fatalf("Since missing: got %v, want ErrNoFact", err)
}
// presence cold-start ⇒ away, score 0.
pres, err := cli.Presence(ctx)
if err != nil {
t.Fatalf("Presence: %v", err)
}
if pres.Bucket != Away || pres.Score != 0 {
t.Fatalf("Presence cold-start = %+v, want away/0", pres)
}
// reminder lifecycle: create → mark fired → re-mark ⇒ ErrReminderState.
rid, err := cli.CreateReminder(ctx, now.Add(time.Hour), `{"text":"wake me 7"}`, "")
if err != nil {
t.Fatalf("CreateReminder: %v", err)
}
if err := cli.MarkReminder(ctx, rid, "fired"); err != nil {
t.Fatalf("MarkReminder fired: %v", err)
}
if err := cli.MarkReminder(ctx, rid, "fired"); !errors.Is(err, ErrReminderState) {
t.Fatalf("MarkReminder twice: got %v, want ErrReminderState", err)
}
// nudge lifecycle: record → resolve acted → resolve again ⇒ ErrNudgeOutcome.
nid, err := cli.RecordNudge(ctx, "water", "voice", "drink", now)
if err != nil {
t.Fatalf("RecordNudge: %v", err)
}
if err := cli.ResolveNudge(ctx, nid, "acted", now); err != nil {
t.Fatalf("ResolveNudge acted: %v", err)
}
if err := cli.ResolveNudge(ctx, nid, "ignored", now); !errors.Is(err, ErrNudgeOutcome) {
t.Fatalf("ResolveNudge twice: got %v, want ErrNudgeOutcome", err)
}
// feedback loop read: RecentOutcomes returns the resolved outcome.
out, err := cli.RecentOutcomes(ctx, "water", 5)
if err != nil {
t.Fatalf("RecentOutcomes: %v", err)
}
if len(out) != 1 || out[0] != "acted" {
t.Fatalf("RecentOutcomes = %v, want [acted]", out)
}
// empty result over the wire is a stable [] not null (server coerces).
if got, err := cli.RecentOutcomes(ctx, "never_fired_rule", 5); err != nil || len(got) != 0 {
t.Fatalf("RecentOutcomes empty = %v err=%v, want []", got, err)
}
}
// TestCaller_Peercred — when the client dials, core sees a Caller with the
// test process's own uid via SO_PEERCRED. This is the seam auth scopes on;
// asserting it's populated today means the future auth layer has its input.
func TestCaller_Peercred(t *testing.T) {
_, srv, cli, _ := newServerWithStore(t)
ctx := context.Background()
// round-trip any call; the server annotates ctx with a Caller on accept.
if _, err := cli.LatestFact(ctx, "nope"); err != nil && !errors.Is(err, ErrNoFact) {
t.Fatalf("LatestFact: %v", err)
}
// introspect the server's view: re-accept a conn manually and read creds.
uc, err := dialRaw(srv.Path())
if err != nil {
t.Fatalf("dialRaw: %v", err)
}
defer uc.Close()
c, ok := peerCaller(uc)
if !ok {
t.Skip("SO_PEERCRED unavailable on this platform; skipping")
}
if c.Uid != int32(os.Getuid()) {
t.Fatalf("peercred uid = %d, want %d", c.Uid, os.Getuid())
}
}
// TestChatViaClient — Chat round-trips over the wire. Uses a custom CoreAPI
// that implements Chat (storeAPI returns an error for it).
func TestChatViaClient(t *testing.T) {
dir := t.TempDir()
sock := filepath.Join(dir, "maven.sock")
chatAPI := &chatTestAPI{}
srv, err := Listen(sock, chatAPI)
if err != nil {
t.Fatalf("listen: %v", err)
}
done := make(chan struct{})
go func() {
_ = srv.Serve()
close(done)
}()
t.Cleanup(func() {
_ = srv.Close()
<-done
})
cli, err := Dial(srv.Path())
if err != nil {
t.Fatalf("dial: %v", err)
}
t.Cleanup(func() { _ = cli.Close() })
reply, err := cli.Chat(context.Background(), "привет")
if err != nil {
t.Fatalf("Chat: %v", err)
}
if reply != "и тебе привет!" {
t.Fatalf("Chat = %q, want %q", reply, "и тебе привет!")
}
}
// chatTestAPI — a minimal CoreAPI that only implements Chat for testing.
// Embeds UnimplementedCoreAPI so every other method fails loudly with
// ErrNotImplemented instead of needing 27 hand-written no-op stubs.
type chatTestAPI struct {
UnimplementedCoreAPI
}
func (a *chatTestAPI) Chat(ctx context.Context, text string) (string, error) {
if text == "привет" {
return "и тебе привет!", nil
}
return "поговорили.", nil
}
// TestDispatch_UnknownMethod — an unknown method over the wire comes back as
// ErrUnknownMethod, not a panic or a dropped conn. The server must stay up
// for the next (legitimate) request on the same conn.
func TestDispatch_UnknownMethod(t *testing.T) {
_, srv, _, _ := newServerWithStore(t)
uc, err := dialRaw(srv.Path())
if err != nil {
t.Fatalf("dialRaw: %v", err)
}
defer uc.Close()
// send garbage method on the raw conn, read back its error, then send a
// real method on the SAME conn to confirm the server survived.
if err := writeFrame(uc, Request{Method: Method("definitely_not_a_method")}); err != nil {
t.Fatalf("writeFrame: %v", err)
}
var resp Response
if err := readFrame(uc, &resp); err != nil {
t.Fatalf("readFrame: %v", err)
}
if resp.Error == nil || !errors.Is(hydrate(resp.Error), ErrUnknownMethod) {
t.Fatalf("unknown method response = %+v, want ErrUnknownMethod", resp.Error)
}
// same conn, legit follow-up: prove the goroutine is still alive.
if err := writeFrame(uc, Request{Method: MethodLatestFact, Params: mustJSON(keyReq{Key: "nope"})}); err != nil {
t.Fatalf("writeFrame follow-up: %v", err)
}
if err := readFrame(uc, &resp); err != nil {
t.Fatalf("readFrame follow-up: %v", err)
}
if resp.Error == nil || !errors.Is(hydrate(resp.Error), ErrNoFact) {
t.Fatalf("follow-up response = %+v, want ErrNoFact", resp.Error)
}
}
// dialRaw — a bare unix conn for tests that want to script the wire directly
// (send an unknown method, follow up on the same conn, inspect framing).
func dialRaw(path string) (net.Conn, error) {
return net.Dial("unix", path)
}
// crashAfterReceiveServer accepts exactly one connection, reads exactly one
// request frame (so, from the client's point of view, the request definitely
// reached the server — a real write could have already committed at this
// point), then closes the connection without ever writing a reply. This is
// the "commit-then-disconnect" scenario the audit finding is about: the
// client cannot tell success from failure from the dropped connection alone.
func crashAfterReceiveServer(t *testing.T, path string) {
t.Helper()
l, err := net.Listen("unix", path)
if err != nil {
t.Fatalf("listen: %v", err)
}
t.Cleanup(func() { _ = l.Close() })
go func() {
conn, err := l.Accept()
if err != nil {
return
}
var req Request
_ = readFrame(conn, &req)
_ = conn.Close() // crash: request received, no reply ever sent
}()
}
// TestClient_MutationNotRetriedOnAmbiguousDisconnect — the audit's core
// finding (Vikunja #269): a write whose reply never arrived (server received
// the frame, then died before replying) must not be silently retried, since
// the original request may have already committed. The client must surface
// ErrAmbiguousOutcome instead of guessing either way.
func TestClient_MutationNotRetriedOnAmbiguousDisconnect(t *testing.T) {
sock := tmpSocket(t)
crashAfterReceiveServer(t, sock)
cli, err := Dial(sock)
if err != nil {
t.Fatalf("dial: %v", err)
}
defer cli.Close()
_, err = cli.WriteFact(context.Background(), WriteFactReq{Key: "k", Value: "v", Source: "test"})
if !errors.Is(err, ErrAmbiguousOutcome) {
t.Fatalf("expected ErrAmbiguousOutcome on commit-then-disconnect, got %v", err)
}
}
// TestClient_ReadRetriedOnAmbiguousDisconnect — the same disconnect timing on
// a read-only method is safe to retry (replaying a read can't double-apply):
// the first connection receives the request and dies without replying, the
// second connection (the client's automatic retry redial) gets a real reply.
// No sleeps: each accepted connection is handled deterministically by index.
func TestClient_ReadRetriedOnAmbiguousDisconnect(t *testing.T) {
sock := tmpSocket(t)
l, err := net.Listen("unix", sock)
if err != nil {
t.Fatalf("listen: %v", err)
}
t.Cleanup(func() { _ = l.Close() })
var n int
go func() {
for {
conn, err := l.Accept()
if err != nil {
return
}
n++
attempt := n
go func(conn net.Conn, attempt int) {
defer conn.Close()
var req Request
if err := readFrame(conn, &req); err != nil {
return
}
if attempt == 1 {
return // crash: request received, no reply ever sent
}
_ = writeFrame(conn, Response{Result: mustJSON(Presence{})})
}(conn, attempt)
}
}()
cli, err := Dial(sock)
if err != nil {
t.Fatalf("dial: %v", err)
}
defer cli.Close()
if _, err := cli.Presence(context.Background()); err != nil {
t.Fatalf("read should have retried past the ambiguous disconnect, got: %v", err)
}
}
func mustJSON(v any) []byte {
b, err := json.Marshal(v)
if err != nil {
panic(err)
}
return b
}
// TestIngestMail_OffUnlessConfigured — with no IngestMailFn set (the default,
// and what an unconfigured core looks like) the method does not exist. A mail
// reader gets a refusal it can act on rather than a silent success.
func TestIngestMail_OffUnlessConfigured(t *testing.T) {
_, _, cli, _ := newServerWithStore(t)
if _, err := cli.IngestMail(context.Background(), IngestMailReq{Mailbox: "INBOX", UID: 1}); !errors.Is(err, ErrUnknownMethod) {
t.Fatalf("IngestMail error = %v, want ErrUnknownMethod", err)
}
}
// TestIngestMail_Hook — when the daemon wires the hook, the message crosses the
// boundary intact and the response comes back.
func TestIngestMail_Hook(t *testing.T) {
_, srv, cli, _ := newServerWithStore(t)
var got IngestMailReq
srv.IngestMailFn = func(_ context.Context, req IngestMailReq) (IngestMailResp, error) {
got = req
return IngestMailResp{TaskIDs: []int64{7}, Created: 1}, nil
}
resp, err := cli.IngestMail(context.Background(), IngestMailReq{
Mailbox: "INBOX", UID: 12, Subject: "Счёт", Body: "Оплатить.", Junk: false,
})
if err != nil {
t.Fatalf("IngestMail: %v", err)
}
if resp.Created != 1 || len(resp.TaskIDs) != 1 || resp.TaskIDs[0] != 7 {
t.Errorf("resp = %+v", resp)
}
if got.UID != 12 || got.Subject != "Счёт" || got.Body != "Оплатить." {
t.Errorf("req across the wire = %+v", got)
}
}
// TestSwapModel_OffUnlessConfigured — no allowlist in the config means the
// daemon never sets the hook, and the method does not exist. That is what "off
// unless configured" looks like at the wire for the model swap (Vikunja #250).
func TestSwapModel_OffUnlessConfigured(t *testing.T) {
_, _, cli, _ := newServerWithStore(t)
if _, err := cli.SwapModel(context.Background(), SwapModelReq{ModelPath: "/m/x.gguf"}); !errors.Is(err, ErrUnknownMethod) {
t.Fatalf("SwapModel error = %v, want ErrUnknownMethod", err)
}
if _, err := cli.ModelStatus(context.Background()); !errors.Is(err, ErrUnknownMethod) {
t.Fatalf("ModelStatus error = %v, want ErrUnknownMethod", err)
}
}
// TestSwapModel_Hook — the request crosses the boundary intact and the reported
// identity comes back. A refusal from the daemon's allowlist arrives as
// ErrForbidden, which is what a caller keys its error message off.
func TestSwapModel_Hook(t *testing.T) {
_, srv, cli, _ := newServerWithStore(t)
var got SwapModelReq
srv.SwapModelFn = func(_ context.Context, req SwapModelReq) (SwapModelResp, error) {
got = req
if req.ModelPath != "/m/allowed.gguf" {
return SwapModelResp{}, fmt.Errorf("%w: not allowlisted", ErrForbidden)
}
return SwapModelResp{Model: "allowed", ModelPath: req.ModelPath, BaseURL: "http://127.0.0.1:9", TookMs: 12}, nil
}
resp, err := cli.SwapModel(context.Background(), SwapModelReq{ModelPath: "/m/allowed.gguf", NCtx: 4096})
if err != nil {
t.Fatalf("SwapModel: %v", err)
}
if resp.Model != "allowed" || resp.TookMs != 12 {
t.Errorf("resp = %+v", resp)
}
if got.NCtx != 4096 {
t.Errorf("req across the wire = %+v", got)
}
if _, err := cli.SwapModel(context.Background(), SwapModelReq{ModelPath: "/etc/shadow"}); !errors.Is(err, ErrForbidden) {
t.Fatalf("swap to a non-allowlisted path = %v; want ErrForbidden", err)
}
}
// The eleven-day bug: Close shut the listener but not the accepted conns, so
// an idle client left serveConn parked in readFrame and wg.Wait never
// returned. mavend deadlocked before `defer st.Close()` could re-encrypt the
// database, and every write since the last clean stop was rolled back on the
// next boot. The client here stays connected and idle on purpose.
func TestCloseReturnsWithAnIdleClientConnected(t *testing.T) {
_, srv, cli, _ := newServerWithStore(t)
// Prove the conn is live and then leave it alone — no cli.Close().
if _, err := cli.RecentFacts(context.Background(), 1); err != nil {
t.Fatalf("warm-up call: %v", err)
}
done := make(chan error, 1)
go func() { done <- srv.Close() }()
select {
case err := <-done:
if err != nil {
t.Fatalf("close: %v", err)
}
// Deliberately shorter than closeGrace: the grace timer is the backstop,
// not the mechanism. Closing the conns is what makes readFrame return, and
// if that regresses this waits out the full grace and fails here.
case <-time.After(closeGrace / 2):
t.Fatal("Close blocked on an idle connection — the shutdown deadlock is back")
}
}