diff --git a/internal/ipc/cancel_test.go b/internal/ipc/cancel_test.go new file mode 100644 index 0000000..bd843ff --- /dev/null +++ b/internal/ipc/cancel_test.go @@ -0,0 +1,119 @@ +package ipc + +import ( + "context" + "errors" + "net" + "path/filepath" + "testing" + "time" +) + +// A cancelled context has to abort a call that is already in flight. It did not +// until V-638: call checked ctx once before sending and then blocked in +// roundtrip with no connection deadline, so a daemon that read the frame and +// never answered parked the caller for as long as the socket stayed open. +// +// The server here is that daemon: it accepts, reads nothing, replies nothing. + +func deafServer(t *testing.T) string { + t.Helper() + sock := filepath.Join(t.TempDir(), "deaf.sock") + ln, err := net.Listen("unix", sock) + if err != nil { + t.Fatalf("listen: %v", err) + } + t.Cleanup(func() { _ = ln.Close() }) + go func() { + for { + conn, err := ln.Accept() + if err != nil { + return + } + // Hold it open and say nothing. Closed by the listener cleanup. + t.Cleanup(func() { _ = conn.Close() }) + } + }() + return sock +} + +func TestClientCancelAbortsAReadInFlight(t *testing.T) { + c, err := Dial(deafServer(t)) + if err != nil { + t.Fatalf("dial: %v", err) + } + defer c.Close() + + ctx, cancel := context.WithCancel(context.Background()) + go func() { + time.Sleep(50 * time.Millisecond) + cancel() + }() + + done := make(chan error, 1) + go func() { + _, err := c.Ping(ctx) + done <- err + }() + + select { + case err := <-done: + // Ping is read-only, so the cancellation is reported as itself rather + // than as an ambiguous mutation. + if !errors.Is(err, context.Canceled) { + t.Errorf("got %v, want context.Canceled", err) + } + case <-time.After(5 * time.Second): + t.Fatal("a cancelled Ping did not return") + } +} + +// A mutation cancelled while awaiting the reply may already have committed, so +// it is ErrAmbiguousOutcome and never a retry. That split is the invariant +// internal/ipc/maperr_test.go's neighbours rest on. +func TestClientCancelLeavesAMutationAmbiguous(t *testing.T) { + c, err := Dial(deafServer(t)) + if err != nil { + t.Fatalf("dial: %v", err) + } + defer c.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + + done := make(chan error, 1) + go func() { + _, err := c.WriteFact(ctx, WriteFactReq{Key: "water", Value: "drank"}) + done <- err + }() + + select { + case err := <-done: + if !errors.Is(err, ErrAmbiguousOutcome) { + t.Errorf("got %v, want ErrAmbiguousOutcome", err) + } + case <-time.After(5 * time.Second): + t.Fatal("a cancelled WriteFact did not return") + } +} + +// The deadline itself, with no cancellation: a call on a context with no +// deadline used to have no bound at all. This one has one and must respect it. +func TestClientDeadlineBoundsACall(t *testing.T) { + c, err := Dial(deafServer(t)) + if err != nil { + t.Fatalf("dial: %v", err) + } + defer c.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + + start := time.Now() + if _, err := c.Ping(ctx); err == nil { + t.Fatal("a deaf server answered a Ping") + } + if elapsed := time.Since(start); elapsed > 3*time.Second { + t.Errorf("Ping took %v, want the context deadline to bound it", elapsed) + } +} diff --git a/internal/ipc/client.go b/internal/ipc/client.go index 5cf1fc9..e2b8c40 100644 --- a/internal/ipc/client.go +++ b/internal/ipc/client.go @@ -29,6 +29,12 @@ type Client struct { mu sync.Mutex } +// defaultCallTimeout bounds a call whose context carries no deadline. It is +// the same 120s internal/voice/client.go settles on: long enough for a model +// call on a cold resident model, short enough that a daemon which stopped +// answering does not park the caller forever. +const defaultCallTimeout = 120 * time.Second + // errWriteLost marks a conn drop while sending the request frame: the request // never reached the server (or the server never saw a complete frame), so // retrying is always safe regardless of method — nothing was applied to @@ -163,16 +169,26 @@ func (c *Client) call(ctx context.Context, m Method, params, result any) error { } var resp Response - err := c.roundtrip(m, raw, &resp) + err := c.roundtrip(ctx, m, raw, &resp) switch { case errors.Is(err, errWriteLost): // The request never left; a duplicate send can't double-apply. // Redial (roundtrip re-dials on a nil conn) and retry exactly once. - err = c.roundtrip(m, raw, &resp) + // Not when the caller has given up — a retry would only be a second + // frame nobody is waiting for. + if ctx.Err() == nil { + err = c.roundtrip(ctx, m, raw, &resp) + } case errors.Is(err, errReadLost): if readOnlyMethods[m] { + if ctx.Err() != nil { + // The caller cancelled the read it was waiting for. Nothing + // was applied, so this is the cancellation and not an + // ambiguity. + return ctx.Err() + } // A duplicate read can't double-apply either — safe to replay. - err = c.roundtrip(m, raw, &resp) + err = c.roundtrip(ctx, m, raw, &resp) } else { // The mutation may have already committed server-side. Do not // retry: report the ambiguity instead of guessing. @@ -199,7 +215,14 @@ func (c *Client) call(ctx context.Context, m Method, params, result any) error { // failure is wrapped in errReadLost (ambiguous — call() only retries it for // read-only methods). Either way a failed conn is dropped so the next call // re-dials clean. Caller holds c.mu. -func (c *Client) roundtrip(m Method, raw json.RawMessage, resp *Response) error { +// +// The connection carries a deadline derived from ctx, falling back to +// defaultCallTimeout, and a watchdog closes it if ctx is cancelled mid-call +// (V-638). Before that a daemon which stopped answering parked the caller for +// as long as the socket stayed open. The watchdog closes the conn rather than +// calling drop, because drop wants c.mu and the caller is holding it — the +// closed socket fails the read, and roundtrip drops it on the way out. +func (c *Client) roundtrip(ctx context.Context, m Method, raw json.RawMessage, resp *Response) error { if c.conn == nil { conn, err := netaddr.Dial(c.addr) if err != nil { @@ -207,6 +230,24 @@ func (c *Client) roundtrip(m Method, raw json.RawMessage, resp *Response) error } c.conn = conn } + conn := c.conn + if dl, ok := ctx.Deadline(); ok { + _ = conn.SetDeadline(dl) + } else { + _ = conn.SetDeadline(time.Now().Add(defaultCallTimeout)) + } + defer conn.SetDeadline(time.Time{}) + + done := make(chan struct{}) + defer close(done) + go func() { + select { + case <-ctx.Done(): + _ = conn.Close() + case <-done: + } + }() + if err := writeFrame(c.conn, Request{Method: m, Params: raw}); err != nil { c.drop() return fmt.Errorf("%w: %v", errWriteLost, err)