rpcclient: support canceling in-flight http requests

diff --git a/rpcclient/infrastructure.go b/rpcclient/infrastructure.go
index e34fc18..694e992 100644
--- a/rpcclient/infrastructure.go
+++ b/rpcclient/infrastructure.go
@@ -98,6 +98,10 @@ const (
 	// defaultHTTPTimeout is the default timeout for an http request, so the
 	// request does not block indefinitely.
 	defaultHTTPTimeout = time.Minute
+
+	// sendPostRequestTries is the number of times to retry failed HTTP POST
+	// requests before giving up.
+	sendPostRequestTries = 10
 )
 
 // jsonRequest holds information about a json request that is used to properly
@@ -766,46 +770,54 @@ out:
 // handleSendPostMessage handles performing the passed HTTP request, reading the
 // result, unmarshalling it, and delivering the unmarshalled result to the
 // provided response channel.
-func (c *Client) handleSendPostMessage(jReq *jsonRequest) {
+func (c *Client) handleSendPostMessage(ctx context.Context, jReq *jsonRequest) {
+	c.sendPostRequestAndRespond(ctx, jReq, sendPostRequestTries)
+}
+
+// sendPostRequestWithRetry performs HTTP POST retries and decodes the response
+// result. It returns the raw transport error so callers can decide how to map
+// shutdown-driven cancellation.
+func sendPostRequestWithRetry(ctx context.Context, jReq *jsonRequest,
+	tries int, httpClient *http.Client, config *ConnConfig,
+	batch bool) ([]byte, error) {
+
 	var (
 		lastErr      error
 		backoff      time.Duration
 		httpResponse *http.Response
+		err          error
 	)
 
-	httpURL, err := c.config.httpURL()
+	httpURL, err := config.httpURL()
 	if err != nil {
-		jReq.responseChan <- &Response{
-			err: fmt.Errorf("failed to parse address %v", err),
-		}
-		return
+		return nil, fmt.Errorf("failed to parse address %v", err)
 	}
 
-	tries := 10
+retryloop:
 	for i := 0; i < tries; i++ {
 		var httpReq *http.Request
 
 		bodyReader := bytes.NewReader(jReq.marshalledJSON)
-		httpReq, err = http.NewRequest("POST", httpURL, bodyReader)
+		httpReq, err = http.NewRequestWithContext(
+			ctx, "POST", httpURL, bodyReader,
+		)
 		if err != nil {
-			jReq.responseChan <- &Response{result: nil, err: err}
-			return
+			return nil, err
 		}
 		httpReq.Close = true
 		httpReq.Header.Set("Content-Type", "application/json")
-		for key, value := range c.config.ExtraHeaders {
+		for key, value := range config.ExtraHeaders {
 			httpReq.Header.Set(key, value)
 		}
 
 		// Configure basic access authorization.
-		user, pass, err := c.config.getAuth()
-		if err != nil {
-			jReq.responseChan <- &Response{result: nil, err: err}
-			return
+		user, pass, authErr := config.getAuth()
+		if authErr != nil {
+			return nil, authErr
 		}
 		httpReq.SetBasicAuth(user, pass)
 
-		httpResponse, err = c.httpClient.Do(httpReq)
+		httpResponse, err = httpClient.Do(httpReq)
 
 		// Quit the retry loop on success or if we can't retry anymore.
 		if err == nil || i == tries-1 {
@@ -830,39 +842,35 @@ func (c *Client) handleSendPostMessage(jReq *jsonRequest) {
 		select {
 		case <-time.After(backoff):
 
-		case <-c.shutdown:
-			return
+		case <-ctx.Done():
+			// Stop retrying as soon as shutdown cancels the request context.
+			err = ctx.Err()
+			break retryloop
 		}
 	}
 	if err != nil {
-		jReq.responseChan <- &Response{err: err}
-		return
+		return nil, err
 	}
 
 	// We still want to return an error if for any reason the response
 	// remains empty.
 	if httpResponse == nil {
-		jReq.responseChan <- &Response{
-			err: fmt.Errorf("invalid http POST response (nil), "+
-				"method: %s, id: %d, last error=%v",
-				jReq.method, jReq.id, lastErr),
-		}
-		return
+		return nil, fmt.Errorf("invalid http POST response (nil), "+
+			"method: %s, id: %d, last error=%v",
+			jReq.method, jReq.id, lastErr)
 	}
 
 	// Read the raw bytes and close the response.
 	respBytes, err := io.ReadAll(httpResponse.Body)
 	httpResponse.Body.Close()
 	if err != nil {
-		err = fmt.Errorf("error reading json reply: %v", err)
-		jReq.responseChan <- &Response{err: err}
-		return
+		return nil, fmt.Errorf("error reading json reply: %w", err)
 	}
 
 	// Try to unmarshal the response as a regular JSON-RPC response.
 	var resp rawResponse
 	var batchResponse json.RawMessage
-	if c.batch {
+	if batch {
 		err = json.Unmarshal(respBytes, &batchResponse)
 	} else {
 		err = json.Unmarshal(respBytes, &resp)
@@ -871,50 +879,70 @@ func (c *Client) handleSendPostMessage(jReq *jsonRequest) {
 		// When the response itself isn't a valid JSON-RPC response
 		// return an error which includes the HTTP status code and raw
 		// response bytes.
-		err = fmt.Errorf("status code: %d, response: %q",
+		return nil, fmt.Errorf("status code: %d, response: %q",
 			httpResponse.StatusCode, string(respBytes))
-		jReq.responseChan <- &Response{err: err}
-		return
 	}
-	var res []byte
-	if c.batch {
-		// errors must be dealt with downstream since a whole request cannot
-		// "error out" other than through the status code error handled above
-		res, err = batchResponse, nil
-	} else {
-		res, err = resp.result()
+
+	if batch {
+		// Errors must be dealt with downstream since a whole request
+		// cannot "error out" other than through the status code error
+		// handled above.
+		return batchResponse, nil
+	}
+
+	return resp.result()
+}
+
+// sendPostRequestAndRespond runs the retrying POST path and sends the final
+// result to the waiting response channel.
+func (c *Client) sendPostRequestAndRespond(ctx context.Context,
+	jReq *jsonRequest, tries int) {
+
+	res, err := sendPostRequestWithRetry(
+		ctx, jReq, tries, c.httpClient, c.config, c.batch,
+	)
+
+	// Preserve the client contract that shutdown-related cancellations surface
+	// as ErrClientShutdown, even when the transport reports context.Canceled.
+	if errors.Is(err, context.Canceled) &&
+		errors.Is(context.Cause(ctx), ErrClientShutdown) {
+
+		err = ErrClientShutdown
+	}
+
+	jReq.responseChan <- &Response{
+		result: res,
+		err:    err,
 	}
-	jReq.responseChan <- &Response{result: res, err: err}
 }
 
 // sendPostHandler handles all outgoing messages when the client is running
 // in HTTP POST mode.  It uses a buffered channel to serialize output messages
 // while allowing the sender to continue running asynchronously.  It must be run
 // as a goroutine.
-func (c *Client) sendPostHandler() {
+func (c *Client) sendPostHandler(ctx context.Context) {
 out:
 	for {
 		// Send any messages ready for send until the shutdown channel
 		// is closed.
 		select {
 		case jReq := <-c.sendPostChan:
-			c.handleSendPostMessage(jReq)
+			c.handleSendPostMessage(ctx, jReq)
 
-		case <-c.shutdown:
+		case <-ctx.Done():
 			break out
 		}
 	}
 
+	err := context.Cause(ctx)
+
 	// Drain any wait channels before exiting so nothing is left waiting
 	// around to send.
 cleanup:
 	for {
 		select {
 		case jReq := <-c.sendPostChan:
-			jReq.responseChan <- &Response{
-				result: nil,
-				err:    ErrClientShutdown,
-			}
+			jReq.responseChan <- &Response{result: nil, err: err}
 
 		default:
 			break cleanup
@@ -1178,8 +1206,13 @@ func (c *Client) start() {
 	// Start the I/O processing handlers depending on whether the client is
 	// in HTTP POST mode or the default websocket mode.
 	if c.config.HTTPPostMode {
+		ctx, cancel := context.WithCancelCause(context.Background())
 		c.wg.Add(1)
-		go c.sendPostHandler()
+		go c.sendPostHandler(ctx)
+		go func() {
+			<-c.shutdown
+			cancel(ErrClientShutdown)
+		}()
 	} else {
 		c.wg.Add(3)
 		go func() {
diff --git a/rpcclient/infrastructure_test.go b/rpcclient/infrastructure_test.go
index 8416b7a..90795b5 100644
--- a/rpcclient/infrastructure_test.go
+++ b/rpcclient/infrastructure_test.go
@@ -1,11 +1,208 @@
 package rpcclient
 
 import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"strings"
+	"sync/atomic"
 	"testing"
+	"time"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
+// postRoundTripFunc adapts a function to implement http.RoundTripper.
+type postRoundTripFunc func(*http.Request) (*http.Response, error)
+
+// RoundTrip invokes the wrapped test transport function.
+func (f postRoundTripFunc) RoundTrip(
+	req *http.Request) (*http.Response, error) {
+
+	return f(req)
+}
+
+// cancelOnReadBody is a test response body that blocks reads until context
+// cancellation is observed.
+type cancelOnReadBody struct {
+	// ctx is the request context that drives cancellation.
+	ctx context.Context
+	// readStarted is closed when the first read call starts.
+	readStarted chan struct{}
+	// stage names the part of the request flow waiting on cancellation.
+	stage string
+}
+
+// Read blocks until the request context is canceled, then returns that error.
+func (b *cancelOnReadBody) Read(_ []byte) (int, error) {
+	select {
+	case <-b.readStarted:
+	default:
+		close(b.readStarted)
+	}
+
+	return 0, waitForRequestContextCancellation(b.ctx, b.stage)
+}
+
+// Close implements io.Closer for the test body.
+func (b *cancelOnReadBody) Close() error {
+	return nil
+}
+
+// newPostModeTestClient builds a minimal HTTP POST-mode client for transport
+// behavior tests.
+func newPostModeTestClient(rt http.RoundTripper) *Client {
+	return &Client{
+		config: &ConnConfig{
+			Host:         "127.0.0.1:8332",
+			User:         "user",
+			Pass:         "pass",
+			DisableTLS:   true,
+			HTTPPostMode: true,
+		},
+		httpClient: &http.Client{
+			Transport: rt,
+		},
+	}
+}
+
+// newPostTestRequest creates a minimal JSON-RPC request used by POST handler
+// tests.
+func newPostTestRequest() *jsonRequest {
+	body := `{"jsonrpc":"1.0","id":1,"method":"getblockcount","params":[]}`
+
+	return &jsonRequest{
+		id:             1,
+		method:         "getblockcount",
+		marshalledJSON: []byte(body),
+		responseChan:   make(chan *Response, 1),
+	}
+}
+
+// waitForRequestContextCancellation bounds shutdown waits so a missing request
+// context propagation becomes a symptomatic test failure instead of a hang.
+func waitForRequestContextCancellation(
+	ctx context.Context, stage string) error {
+
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+
+	case <-time.After(100 * time.Millisecond):
+		return fmt.Errorf("request context was not canceled during %s",
+			stage)
+	}
+}
+
+// sendPostShutdownScenario describes one shutdown path that should fail if the
+// request stops honoring shutdown cancellation.
+type sendPostShutdownScenario struct {
+	// name is the subtest name for the shutdown path.
+	name string
+
+	// tries is the retry count passed to the POST send helper under test.
+	tries int
+
+	// newClient builds a POST-mode client whose transport triggers the
+	// shutdown path for this scenario.
+	newClient func(context.CancelCauseFunc, *int32) *Client
+
+	// wantAttempts is the expected number of transport attempts before the
+	// scenario terminates.
+	wantAttempts int32
+
+	// wantErrContains is an optional substring that must appear in the raw
+	// helper error for this scenario.
+	wantErrContains string
+}
+
+// sendPostShutdownScenarios enumerates the shutdown-triggered regression
+// scenarios shared by the helper-level and wrapper-level POST tests.
+var sendPostShutdownScenarios = []sendPostShutdownScenario{
+	{
+		name:  "during_retry_backoff",
+		tries: 2,
+		newClient: func(cancel context.CancelCauseFunc,
+			attempts *int32) *Client {
+
+			return newPostModeTestClient(postRoundTripFunc(
+				func(*http.Request) (*http.Response, error) {
+					if atomic.AddInt32(attempts, 1) == 1 {
+						cancel(ErrClientShutdown)
+					}
+
+					return nil, errors.New(
+						"transient transport error",
+					)
+				},
+			))
+		},
+		wantAttempts: 1,
+	},
+	{
+		name:  "on_final_retry",
+		tries: 2,
+		newClient: func(cancel context.CancelCauseFunc,
+			attempts *int32) *Client {
+
+			return newPostModeTestClient(postRoundTripFunc(
+				func(req *http.Request) (*http.Response, error) {
+					current := atomic.AddInt32(attempts, 1)
+					if current == 1 {
+						return nil, errors.New(
+							"transient transport error",
+						)
+					}
+
+					// This keeps the case tied to request-context
+					// propagation instead of injecting context.Canceled
+					// directly from the fake transport.
+					cancel(ErrClientShutdown)
+					return nil, waitForRequestContextCancellation(
+						req.Context(), "final retry",
+					)
+				},
+			))
+		},
+		wantAttempts: 2,
+	},
+	{
+		name:  "during_body_read",
+		tries: 1,
+		newClient: func(cancel context.CancelCauseFunc,
+			attempts *int32) *Client {
+
+			readStarted := make(chan struct{})
+			go func() {
+				<-readStarted
+				cancel(ErrClientShutdown)
+			}()
+
+			return newPostModeTestClient(postRoundTripFunc(
+				func(req *http.Request) (*http.Response, error) {
+					atomic.AddInt32(attempts, 1)
+
+					return &http.Response{
+						StatusCode: http.StatusOK,
+						Header:     make(http.Header),
+						Body: &cancelOnReadBody{
+							ctx:         req.Context(),
+							readStarted: readStarted,
+							stage:       "body read",
+						},
+					}, nil
+				},
+			))
+		},
+		wantAttempts:    1,
+		wantErrContains: "error reading json reply",
+	},
+}
+
 // TestParseAddressString checks different variation of supported and
 // unsupported addresses.
 func TestParseAddressString(t *testing.T) {
@@ -93,18 +290,191 @@ func TestParseAddressString(t *testing.T) {
 	}
 
 	for _, tc := range testCases {
-		tc := tc
-
 		t.Run(tc.name, func(t *testing.T) {
 			addr, err := ParseAddressString(tc.addressString)
 			if tc.expErrStr != "" {
 				require.Error(t, err)
 				require.Contains(t, err.Error(), tc.expErrStr)
-				return
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, tc.expNetwork, addr.Network())
+				require.Equal(t, tc.expAddress, addr.String())
+			}
+		})
+	}
+}
+
+// TestSendPostRequestWithRetrySuccess ensures that
+// sendPostRequestWithRetry returns a decoded result and no error on
+// a successful response.
+func TestSendPostRequestWithRetrySuccess(t *testing.T) {
+	client := newPostModeTestClient(postRoundTripFunc(
+		func(*http.Request) (*http.Response, error) {
+			return &http.Response{
+				StatusCode: http.StatusOK,
+				Header:     make(http.Header),
+				Body: io.NopCloser(strings.NewReader(
+					`{"result":1,"error":null,"id":1}`,
+				)),
+			}, nil
+		},
+	))
+	jReq := newPostTestRequest()
+
+	result, err := sendPostRequestWithRetry(
+		context.Background(), jReq, 1, client.httpClient, client.config,
+		false,
+	)
+	require.NoError(t, err)
+	require.Equal(t, []byte("1"), result)
+}
+
+// TestSendPostRequestWithRetryShutdown keeps the shutdown regression cases in
+// one table while preserving a distinct symptomatic failure for each path.
+func TestSendPostRequestWithRetryShutdown(t *testing.T) {
+	for _, tc := range sendPostShutdownScenarios {
+		t.Run(tc.name, func(t *testing.T) {
+			var attempts int32
+			ctx, cancel := context.WithCancelCause(context.Background())
+			client := tc.newClient(cancel, &attempts)
+			jReq := newPostTestRequest()
+
+			result, err := sendPostRequestWithRetry(
+				ctx, jReq, tc.tries, client.httpClient, client.config,
+				false,
+			)
+			require.Nil(t, result)
+			require.ErrorIs(t, err, context.Canceled)
+			if tc.wantErrContains != "" {
+				require.ErrorContains(t, err, tc.wantErrContains)
+			}
+			require.EqualValues(t, tc.wantAttempts,
+				atomic.LoadInt32(&attempts))
+		})
+	}
+}
+
+// TestHTTPPostShutdownInterruptsPendingRequest ensures that a client operating
+// in HTTP POST mode can interrupt an in-flight request during shutdown.
+func TestHTTPPostShutdownInterruptsPendingRequest(t *testing.T) {
+	t.Parallel()
+
+	// Start a local TCP listener that accepts exactly one HTTP request and
+	// then blocks until the client side closes the connection.
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+
+	// requestAccepted signals when the test server has accepted the
+	// client's connection.
+	requestAccepted := make(chan struct{})
+
+	// serverDone signals when the server goroutine has exited.
+	serverDone := make(chan struct{})
+
+	// Run a minimum server goroutine. It accepts one connection and drains
+	// the request stream without replying so the client request stays in
+	// flight.
+	go func() {
+		defer close(serverDone)
+
+		conn, err := listener.Accept()
+		if err != nil {
+			return
+		}
+		defer func() {
+			err := conn.Close()
+			assert.NoError(t, err)
+		}()
+
+		close(requestAccepted)
+
+		_, _ = io.Copy(io.Discard, conn)
+	}()
+
+	// Ensure the listener is closed and the server goroutine exits.
+	t.Cleanup(func() {
+		err := listener.Close()
+		require.NoError(t, err)
+		<-serverDone
+	})
+
+	// Configure a POST-mode client against the local listener.
+	connCfg := &ConnConfig{
+		Host:         listener.Addr().String(),
+		User:         "user",
+		Pass:         "pass",
+		DisableTLS:   true,
+		HTTPPostMode: true,
+	}
+
+	// Start the client and register cleanup for idempotent shutdown.
+	client, err := New(connCfg, nil)
+	require.NoError(t, err)
+	t.Cleanup(client.Shutdown)
+
+	// Launch one async request that should remain pending until shutdown.
+	future := client.GetBlockCountAsync()
+
+	// Ensure the server sees the request before we initiate shutdown.
+	select {
+	case <-requestAccepted:
+
+	case <-time.After(2 * time.Second):
+		t.Fatalf("server did not accept client connection")
+	}
+
+	// The request should remain pending until shutdown is requested.
+	select {
+	case <-future:
+		t.Fatalf("expected request to remain pending until shutdown")
+
+	case <-time.After(100 * time.Millisecond):
+	}
+
+	client.Shutdown()
+
+	waitDone := make(chan struct{})
+	go func() {
+		client.WaitForShutdown()
+		close(waitDone)
+	}()
+
+	// Wait for shutdown to complete before asserting the final error.
+	select {
+	case <-waitDone:
+
+	case <-time.After(5 * time.Second):
+		t.Fatalf("client shutdown did not complete")
+	}
+
+	result, err := future.Receive()
+	require.Zero(t, result)
+	require.ErrorContains(t, err, ErrClientShutdown.Error())
+}
+
+// TestSendPostRequestAndRespondShutdown reuses the helper-level shutdown cases
+// to verify the client-facing contract: each one must surface
+// ErrClientShutdown on the response channel.
+func TestSendPostRequestAndRespondShutdown(t *testing.T) {
+	for _, tc := range sendPostShutdownScenarios {
+		t.Run(tc.name, func(t *testing.T) {
+			var attempts int32
+			ctx, cancel := context.WithCancelCause(context.Background())
+			client := tc.newClient(cancel, &attempts)
+			jReq := newPostTestRequest()
+
+			go client.sendPostRequestAndRespond(ctx, jReq, tc.tries)
+
+			select {
+			case resp := <-jReq.responseChan:
+				require.ErrorIs(t, resp.err, ErrClientShutdown)
+				require.Nil(t, resp.result)
+			case <-time.After(2 * time.Second):
+				t.Fatal("timed out waiting for response")
 			}
-			require.NoError(t, err)
-			require.Equal(t, tc.expNetwork, addr.Network())
-			require.Equal(t, tc.expAddress, addr.String())
+
+			require.EqualValues(t, tc.wantAttempts,
+				atomic.LoadInt32(&attempts))
 		})
 	}
 }