Propagate RPC cancellation to daemon requests

Stop long-running daemon operations from running under context.Background by\nthreading a request-scoped context from handleConn into dispatch. The daemon\nnow cancels in-flight handlers when the client socket goes away, and the RPC\nclient closes its Unix connection when the caller context is canceled so that\ninterrupts actually reach the daemon boundary.\n\nAdd regression coverage for both sides of the path: canceled dispatch calls,\nclient disconnects during handleConn, watcher EOF cancellation, and context\ncancellation without an RPC deadline.\n\nValidated with GOCACHE=/tmp/banger-gocache go test ./... and\nGOCACHE=/tmp/banger-gocache make build.
This commit is contained in:
Thales Maciel 2026-03-16 18:28:33 -03:00
parent ebb68c3126
commit ccba07ec68
No known key found for this signature in database
GPG key ID: 33112E6833C34679
4 changed files with 220 additions and 16 deletions

View file

@ -57,11 +57,22 @@ func DecodeParams[T any](req Request) (T, error) {
func Call[T any](ctx context.Context, socketPath, method string, params any) (T, error) {
var zero T
conn, err := net.DialTimeout("unix", socketPath, 2*time.Second)
dialer := &net.Dialer{Timeout: 2 * time.Second}
conn, err := dialer.DialContext(ctx, "unix", socketPath)
if err != nil {
return zero, err
}
defer conn.Close()
done := make(chan struct{})
defer close(done)
go func() {
select {
case <-ctx.Done():
_ = conn.SetDeadline(time.Now())
_ = conn.Close()
case <-done:
}
}()
if deadline, ok := ctx.Deadline(); ok {
_ = conn.SetDeadline(deadline)
@ -77,11 +88,17 @@ func Call[T any](ctx context.Context, socketPath, method string, params any) (T,
}
if err := json.NewEncoder(conn).Encode(request); err != nil {
if ctx.Err() != nil {
return zero, ctx.Err()
}
return zero, err
}
var response Response
if err := json.NewDecoder(bufio.NewReader(conn)).Decode(&response); err != nil {
if ctx.Err() != nil {
return zero, ctx.Err()
}
return zero, err
}
if !response.OK {

View file

@ -128,6 +128,32 @@ func TestCallHonorsContextDeadline(t *testing.T) {
}
}
func TestCallHonorsContextCancellationWithoutDeadline(t *testing.T) {
t.Parallel()
socketPath, cleanup := serveRPCOnce(t, func(conn net.Conn) {
defer conn.Close()
var req Request
if err := json.NewDecoder(bufio.NewReader(conn)).Decode(&req); err != nil {
t.Fatalf("decode request: %v", err)
}
var buf [1]byte
_, _ = conn.Read(buf[:])
})
defer cleanup()
ctx, cancel := context.WithCancel(context.Background())
go func() {
time.Sleep(20 * time.Millisecond)
cancel()
}()
_, err := Call[map[string]string](ctx, socketPath, "ping", nil)
if !errors.Is(err, context.Canceled) {
t.Fatalf("Call() error = %v, want context canceled", err)
}
}
func TestWaitForSocket(t *testing.T) {
t.Parallel()