an ipc call can be cancelled and always has a deadline (V-638)
roundtrip set no connection deadline and call checked the context once, before sending, so a daemon that read the frame and stopped answering parked the caller for as long as the socket stayed open. The deadline comes from the caller's context, falling back to 120s, the convention internal/voice/client.go already had. A watchdog closes the conn on ctx.Done(); it closes rather than calling drop, because drop wants c.mu and call is holding it. The retry split is unchanged and now also declines a retry the caller has stopped waiting for. A cancelled mutation stays ErrAmbiguousOutcome, because it may have committed. A cancelled read reports the cancellation. Three tests against a server that accepts and never answers. There was no test for this, which is why it went unnoticed.
This commit is contained in:
+45
-4
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user