amadan

carlos / platform Public

WIP: a Stripe request cannot outlive its connection

One connection per request, established before the last check, never reused,
kept alive or retried, HTTP/1.1, no proxy. The check reads the caller's send
deadline and the context's deadline on the wall clock, so a suspended sender
wakes to a closed window. A request held past the provider's idle limit is
written into a closed connection and never processed.

Not gated, not reviewed. The provider's idle limit is not measured yet.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude 38cf9ff5fdef1aa647318fe2e39aaf3c029aa83d parent a78d598
4 files changed, +359 −2
  • internal/billingstripe/client.go +10 −1
  • internal/billingstripe/transport.go +177 −0
  • internal/billingstripe/transport_test.go +166 −0
  • internal/console/billingquantity.go +6 −1
diff --git a/internal/billingstripe/client.go b/internal/billingstripe/client.go
index 22d8696e1..8fdebcb0e 100644
--- a/internal/billingstripe/client.go
+++ b/internal/billingstripe/client.go
@@ -56,6 +56,9 @@ const (
CodeAPIError = "api_error"
CodeBrokenPagination = "broken_pagination"
CodeTooManyRecords = "too_many_records"
+ // CodeSendWindowClosed: the caller's send deadline (WithSendBy) had
+ // passed by the time a connection was ready. Nothing was written.
+ CodeSendWindowClosed = "send_window_closed"
)
// Error is the only error shape this package returns. Its Error() string is
@@ -128,7 +131,10 @@ func NewClient(secretKey string, opts ...Option) (*Client, error) {
secretKey: secretKey,
baseURL: defaultBaseURL,
httpClient: &http.Client{
- Timeout: requestTimeout,
+ // One connection per request, checked after it is established
+ // and never reused, kept alive or retried: transport.go.
+ Transport: newOneShotTransport(),
+ Timeout: requestTimeout,
// Never follow a redirect: Stripe's API does not issue them, and
// following one would replay the Authorization header (and thus
// the restricted key) against a host we did not choose. Treat any
@@ -192,6 +198,9 @@ func (c *Client) do(ctx context.Context, method, path string, form url.Values, i
resp, err := c.httpClient.Do(req)
if err != nil {
+ if errors.Is(err, errSendWindowClosed) {
+ return &Error{Code: CodeSendWindowClosed}
+ }
if ctx.Err() != nil || isTimeout(err) {
return &Error{Code: CodeTimeout}
}
diff --git a/internal/billingstripe/transport.go b/internal/billingstripe/transport.go
new file mode 100644
index 000000000..d46685093
--- /dev/null
+++ b/internal/billingstripe/transport.go
@@ -0,0 +1,177 @@
+package billingstripe
+
+import (
+ "bufio"
+ "context"
+ "crypto/tls"
+ "errors"
+ "io"
+ "net"
+ "net/http"
+ "time"
+)
+
+// The transport: a request cannot outlive its connection.
+//
+// Why. A caller decides it may send (it holds a lease, its operation is
+// still wanted) and then sends. A process can be frozen between the two:
+// stopped, or its whole machine suspended. When it wakes it sends a request
+// that was decided hours or days ago. Nothing the caller checked beforehand
+// can stop that, and no parameter of a Stripe request makes Stripe refuse
+// it for being old: there are no conditional writes, and an idempotency key
+// is forgotten after a day (see docs/superpowers/plans/
+// 2026-10-01-late-request-fencing.md).
+//
+// What can refuse it is the connection it travels on. Every server closes a
+// connection that sits idle. So the request is built to leave in exactly
+// one way:
+//
+// 1. the connection is established first (TCP and TLS);
+// 2. then comes the caller's last check, the send deadline carried by
+// WithSendBy, against the wall clock;
+// 3. then the request is written, on that connection and no other.
+//
+// A process frozen after step 2 holds an established, idle connection for
+// as long as it is frozen. Past the provider's idle limit the provider has
+// closed it, and what the process writes on waking is answered with a reset
+// and never processed. A process frozen before step 2 wakes to a check
+// that fails.
+//
+// For that to be the only way out, the transport gives up everything that
+// would let a request travel some other way:
+//
+// - one connection per request, never reused, closed after the response;
+// - no TCP keepalive, which would let the kernel keep an idle connection
+// looking alive to a middlebox while the process is stopped;
+// - no automatic retry: a failed write is an error, never a re-dial (the
+// standard transport re-sends any request carrying an Idempotency-Key
+// header when a reused connection turns out to be closed);
+// - HTTP/1.1, so "one request per connection" is literally true;
+// - no proxy: an intermediary that holds connections open would be the
+// hop where the argument fails, so the environment's proxy settings are
+// not read.
+//
+// The bound this gives is: a request reaches the provider no later than its
+// check plus the provider's idle limit, or never. The idle limit is the
+// provider's behaviour, not this code's. Stripe does not document it; the
+// sandbox test measures it.
+type oneShotTransport struct {
+ dialer *net.Dialer
+ // afterConnect runs once the connection is established and before the
+ // last check. Tests use it to hold a request the way a freeze would.
+ afterConnect func()
+ // wallNow is the wall clock the last check reads. A test moves it the
+ // way a suspended machine's clock moves on waking: forward, with the
+ // monotonic clock (and so every context deadline) standing still.
+ wallNow func() time.Time
+}
+
+// errSendWindowClosed: the caller's send deadline had passed by the time a
+// connection was ready. Nothing was written.
+var errSendWindowClosed = errors.New("billingstripe: send window closed")
+
+type sendByKey struct{}
+
+// WithSendBy carries the wall-clock instant after which a request made
+// with ctx must not be written: the end of whatever lease entitles the
+// caller to send, less its clock tolerance. The transport checks it after
+// the connection is established and immediately before the write.
+func WithSendBy(ctx context.Context, t time.Time) context.Context {
+ return context.WithValue(ctx, sendByKey{}, t)
+}
+
+func sendByFrom(ctx context.Context) (time.Time, bool) {
+ t, ok := ctx.Value(sendByKey{}).(time.Time)
+ return t, ok && !t.IsZero()
+}
+
+func newOneShotTransport() *oneShotTransport {
+ return &oneShotTransport{dialer: &net.Dialer{Timeout: 10 * time.Second, KeepAlive: -1}}
+}
+
+func (t *oneShotTransport) RoundTrip(req *http.Request) (*http.Response, error) {
+ ctx := req.Context()
+ host, port := req.URL.Hostname(), req.URL.Port()
+ secure := req.URL.Scheme == "https"
+ if port == "" {
+ port = "80"
+ if secure {
+ port = "443"
+ }
+ }
+ raw, err := t.dialer.DialContext(ctx, "tcp", net.JoinHostPort(host, port))
+ if err != nil {
+ return nil, err
+ }
+ var conn net.Conn = raw
+ if secure {
+ tc := tls.Client(raw, &tls.Config{ServerName: host, NextProtos: []string{"http/1.1"}, MinVersion: tls.VersionTLS12})
+ if err := tc.HandshakeContext(ctx); err != nil {
+ raw.Close()
+ return nil, err
+ }
+ conn = tc
+ }
+ // A cancelled or expired context closes the connection, which unblocks
+ // any read or write in progress.
+ stop := context.AfterFunc(ctx, func() { conn.Close() })
+ closeConn := func() {
+ stop()
+ conn.Close()
+ }
+ if hook := t.afterConnect; hook != nil {
+ hook()
+ }
+ // The last check, on an established connection, immediately before the
+ // request is written. Two deadlines, both read on the wall clock: the
+ // caller's (WithSendBy), and the context's own. A context deadline
+ // normally fires by the monotonic clock, which stands still while a
+ // machine is suspended, so a suspended sender would wake with its
+ // context still alive. Its wall-clock reading has passed all the same,
+ // and that is what is compared here. Every request this package makes
+ // carries a context deadline (the client's timeout), so every request
+ // is checked.
+ now := time.Now
+ if t.wallNow != nil {
+ now = t.wallNow
+ }
+ wall := now().Round(0)
+ if by, ok := sendByFrom(ctx); ok && !wall.Before(by.Round(0)) {
+ closeConn()
+ return nil, errSendWindowClosed
+ }
+ if dl, ok := ctx.Deadline(); ok && !wall.Before(dl.Round(0)) {
+ closeConn()
+ return nil, errSendWindowClosed
+ }
+ if err := ctx.Err(); err != nil {
+ closeConn()
+ return nil, err
+ }
+ out := req.Clone(ctx)
+ out.Close = true
+ if err := out.Write(conn); err != nil {
+ closeConn()
+ return nil, err
+ }
+ resp, err := http.ReadResponse(bufio.NewReader(conn), out)
+ if err != nil {
+ closeConn()
+ return nil, err
+ }
+ resp.Body = &closingBody{ReadCloser: resp.Body, done: closeConn}
+ return resp, nil
+}
+
+// closingBody closes the connection when the response body is closed: the
+// connection carried one request and carries nothing else.
+type closingBody struct {
+ io.ReadCloser
+ done func()
+}
+
+func (b *closingBody) Close() error {
+ err := b.ReadCloser.Close()
+ b.done()
+ return err
+}
diff --git a/internal/billingstripe/transport_test.go b/internal/billingstripe/transport_test.go
new file mode 100644
index 000000000..9c3172dca
--- /dev/null
+++ b/internal/billingstripe/transport_test.go
@@ -0,0 +1,166 @@
+package billingstripe
+
+import (
+ "context"
+ "errors"
+ "net"
+ "net/http"
+ "net/http/httptest"
+ "sync/atomic"
+ "testing"
+ "time"
+)
+
+// connCountingServer counts connections opened and requests handled.
+func connCountingServer(t *testing.T, idle time.Duration) (*httptest.Server, *atomic.Int64, *atomic.Int64) {
+ t.Helper()
+ var conns, handled atomic.Int64
+ srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ handled.Add(1)
+ w.Header().Set("Content-Type", "application/json")
+ _, _ = w.Write([]byte(`{"id":"si_1","quantity":11,"price":{"id":"price_1"}}`))
+ }))
+ // A connection that has sent nothing for this long is closed: the
+ // provider's idle limit, made short.
+ srv.Config.ReadHeaderTimeout = idle
+ srv.Config.ConnState = func(_ net.Conn, s http.ConnState) {
+ if s == http.StateNew {
+ conns.Add(1)
+ }
+ }
+ srv.Start()
+ t.Cleanup(srv.Close)
+ return srv, &conns, &handled
+}
+
+func holdAfterConnect(c *Client, hold func()) {
+ c.httpClient.Transport.(*oneShotTransport).afterConnect = hold
+}
+
+// The mechanism the late-request design rests on. A sender is frozen after
+// its last check, with its connection already established. The freeze
+// outlasts the provider's idle limit, so the provider has closed the
+// connection. The request the sender writes on waking is never processed,
+// and nothing sends it again on a new connection.
+func TestRequestHeldPastTheIdleLimitIsNeverProcessed(t *testing.T) {
+ srv, conns, handled := connCountingServer(t, 150*time.Millisecond)
+ c := testClient(t, srv.URL)
+ holdAfterConnect(c, func() { time.Sleep(600 * time.Millisecond) })
+ // The check passes: the deadline is far off. Only the provider's idle
+ // limit stands between the frozen request and its execution.
+ ctx := WithSendBy(context.Background(), time.Now().Add(time.Hour))
+ _, err := c.UpdateQuantityWith(ctx, "si_1", 11, ProrationAlwaysInvoice, time.Time{}, "held-key")
+ if err == nil {
+ t.Fatal("a request written on a connection the provider had closed was answered")
+ }
+ // Give a stray retry time to show itself.
+ time.Sleep(300 * time.Millisecond)
+ if n := handled.Load(); n != 0 {
+ t.Fatalf("the provider processed %d request(s) from the held sender", n)
+ }
+ if n := conns.Load(); n != 1 {
+ t.Fatalf("%d connections were opened: the request was re-sent on a new one", n)
+ }
+}
+
+// A sender frozen BEFORE its last check wakes to a check that fails. The
+// check is made after the connection is established and against the wall
+// clock, and nothing is written.
+func TestRequestIsNotWrittenAfterItsSendDeadline(t *testing.T) {
+ srv, conns, handled := connCountingServer(t, 5*time.Second)
+ c := testClient(t, srv.URL)
+ holdAfterConnect(c, func() { time.Sleep(250 * time.Millisecond) })
+ ctx := WithSendBy(context.Background(), time.Now().Add(100*time.Millisecond))
+ _, err := c.UpdateQuantityWith(ctx, "si_1", 11, ProrationAlwaysInvoice, time.Time{}, "late-key")
+ var e *Error
+ if !errors.As(err, &e) || e.Code != CodeSendWindowClosed {
+ t.Fatalf("err = %v, want %s", err, CodeSendWindowClosed)
+ }
+ time.Sleep(100 * time.Millisecond)
+ if handled.Load() != 0 || conns.Load() != 1 {
+ t.Fatalf("handled %d on %d connection(s): a request past its deadline was written", handled.Load(), conns.Load())
+ }
+ // Inside the deadline the same request goes through.
+ holdAfterConnect(c, nil)
+ ctx = WithSendBy(context.Background(), time.Now().Add(time.Minute))
+ if _, err := c.UpdateQuantityWith(ctx, "si_1", 11, ProrationAlwaysInvoice, time.Time{}, "in-time-key"); err != nil {
+ t.Fatalf("inside the deadline: %v", err)
+ }
+ if handled.Load() != 1 {
+ t.Fatalf("handled %d, want 1", handled.Load())
+ }
+}
+
+// One connection carries one request. A request never travels on a
+// connection that existed before its own check.
+func TestEveryRequestHasItsOwnConnection(t *testing.T) {
+ srv, conns, handled := connCountingServer(t, 5*time.Second)
+ c := testClient(t, srv.URL)
+ for i := 0; i < 3; i++ {
+ if _, err := c.UpdateQuantityWith(context.Background(), "si_1", 11, ProrationAlwaysInvoice, time.Time{}, "key"); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if conns.Load() != 3 || handled.Load() != 3 {
+ t.Fatalf("%d requests on %d connections, want 3 on 3", handled.Load(), conns.Load())
+ }
+}
+
+// A context that ends while the request is held closes the connection: the
+// held request is not written afterwards.
+func TestCancelledRequestIsNotWritten(t *testing.T) {
+ srv, _, handled := connCountingServer(t, 5*time.Second)
+ c := testClient(t, srv.URL)
+ ctx, cancel := context.WithCancel(context.Background())
+ holdAfterConnect(c, func() {
+ cancel()
+ time.Sleep(50 * time.Millisecond)
+ })
+ if _, err := c.UpdateQuantityWith(ctx, "si_1", 11, ProrationAlwaysInvoice, time.Time{}, "cancelled-key"); err == nil {
+ t.Fatal("a cancelled request was answered")
+ }
+ time.Sleep(100 * time.Millisecond)
+ if handled.Load() != 0 {
+ t.Fatalf("a cancelled request was processed %d time(s)", handled.Load())
+ }
+}
+
+// The transport is the client's only way out: the environment's proxy
+// settings are never read, and the dialer sends no keepalive.
+func TestTransportTakesNoProxyAndSendsNoKeepalive(t *testing.T) {
+ c := testClient(t, "http://127.0.0.1:1")
+ tr, ok := c.httpClient.Transport.(*oneShotTransport)
+ if !ok {
+ t.Fatalf("the client's transport is %T", c.httpClient.Transport)
+ }
+ if tr.dialer.KeepAlive >= 0 {
+ t.Fatalf("keepalive %v: an idle connection would be kept looking alive", tr.dialer.KeepAlive)
+ }
+}
+
+// A suspended machine: the wall clock has moved on and the monotonic clock
+// has not, so the request's context is still alive when the sender wakes.
+// The check reads the context's deadline on the wall clock and does not
+// write. No caller has to ask for this: the client's own timeout gives
+// every request a deadline.
+func TestSuspendedSenderWakesToAClosedSendWindow(t *testing.T) {
+ srv, _, handled := connCountingServer(t, 5*time.Second)
+ c := testClient(t, srv.URL)
+ tr := c.httpClient.Transport.(*oneShotTransport)
+ var slept time.Duration
+ tr.wallNow = func() time.Time { return time.Now().Add(slept) }
+ tr.afterConnect = func() { slept = 2 * requestTimeout }
+ _, err := c.UpdateQuantityWith(context.Background(), "si_1", 11, ProrationAlwaysInvoice, time.Time{}, "suspended-key")
+ var e *Error
+ if !errors.As(err, &e) || e.Code != CodeSendWindowClosed {
+ t.Fatalf("err = %v, want %s", err, CodeSendWindowClosed)
+ }
+ if handled.Load() != 0 {
+ t.Fatal("a sender that slept through its deadline wrote its request")
+ }
+ // Awake the whole time, the same request goes through.
+ slept, tr.afterConnect = 0, nil
+ if _, err := c.UpdateQuantityWith(context.Background(), "si_1", 11, ProrationAlwaysInvoice, time.Time{}, "awake-key"); err != nil {
+ t.Fatal(err)
+ }
+}
diff --git a/internal/console/billingquantity.go b/internal/console/billingquantity.go
index 0f71bbfdc..15906c031 100644
--- a/internal/console/billingquantity.go
+++ b/internal/console/billingquantity.go
@@ -1055,7 +1055,12 @@ func (s *Server) sendQuantityOperation(ctx context.Context, state billingState,
// to be sent. Either way this invocation sends nothing.
return quantityStep{done: true}, nil
}
- sendCtx, cancel := context.WithTimeout(ctx, until.Add(-billingClockTolerance).Sub(claimed))
+ // The request must not be written after the lease's margin begins. The
+ // transport makes that its last check, after the connection is
+ // established (billingstripe, transport.go), so a sender frozen here
+ // cannot send late on a connection of its own making.
+ window := until.Add(-billingClockTolerance).Sub(claimed)
+ sendCtx, cancel := context.WithTimeout(billingstripe.WithSendBy(ctx, time.Now().Add(window)), window)
updated, err := b.client.UpdateQuantityWith(sendCtx, op.Item, op.Target, op.Behavior, time.Unix(op.CreatedAt, 0), op.Key)
cancel()
if err != nil {