amadan

carlos / platform Public

db door: keep the write deadline through net/http's final flush

The handler cleared its write deadline on the way out, but net/http
flushes the last buffered bytes and the chunked terminator in
finishRequest after the handler returns, and clears the deadline itself
only after that. So the final flush ran with no deadline, and a client
that stopped reading near the end held the connection forever, after the
stream's slot was already released.

The handler no longer clears the deadline. It renews it and flushes
explicitly before its deferred slot release, so the tail is written or
cut under the deadline, and what finishRequest has left (the
terminator) is still bounded. A test records every write and deadline
change on the server connection and fails if any byte is written with
the deadline cleared.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude f300cd27200f67d77f1f05cd5288a8929602373a parent a93f8f6
2 files changed, +116 −4
  • internal/console/apidb.go +18 −4
  • internal/console/apidb_test.go +98 −0
diff --git a/internal/console/apidb.go b/internal/console/apidb.go
index 5de0cd0d7..bf1a783d0 100644
--- a/internal/console/apidb.go
+++ b/internal/console/apidb.go
@@ -337,12 +337,26 @@ func (s *Server) handleAPIDBGet(w http.ResponseWriter, r *http.Request) {
status = http.StatusPartialContent
}
rw := http.NewResponseController(w)
- // The server sets no WriteTimeout, so a deadline left here would carry
- // over to the next request on a kept-alive connection. Clear it.
- defer rw.SetWriteDeadline(time.Time{})
+ // The deadline is never cleared here. net/http flushes what is still
+ // buffered (the last bytes, the chunked terminator) in finishRequest,
+ // after this handler returns, and clears the write deadline itself
+ // only after that flush. Clearing it on the way out would leave that
+ // flush unbounded, holding the connection for as long as a client that
+ // stopped reading near the end cares to wait.
_ = rw.SetWriteDeadline(time.Now().Add(idle))
w.WriteHeader(status)
- if err := dbStream(w, rw, rc, watchdog, idle); err != nil {
+ err = dbStream(w, rw, rc, watchdog, idle)
+ // Flush the tail now, under a fresh deadline, so it is done (or cut)
+ // before the deferred release gives this stream's slot back. What
+ // finishRequest has left after this is the chunked terminator, still
+ // under the same deadline. On a failed read the deadline is renewed
+ // too, so the client gets its truncated body promptly (and resumes)
+ // rather than a connection cut by a deadline the stall used up.
+ _ = rw.SetWriteDeadline(time.Now().Add(idle))
+ if ferr := rw.Flush(); err == nil {
+ err = ferr
+ }
+ if err != nil {
// The status is already on the wire. The client's resumable reader
// sees a short body against the listed size and reopens at the
// offset it reached.
diff --git a/internal/console/apidb_test.go b/internal/console/apidb_test.go
index 334f83b63..bf7501a20 100644
--- a/internal/console/apidb_test.go
+++ b/internal/console/apidb_test.go
@@ -7,10 +7,13 @@ import (
"errors"
"fmt"
"io"
+ "net"
"net/http"
+ "net/http/httptest"
"os"
"path/filepath"
"strings"
+ "sync"
"sync/atomic"
"testing"
"time"
@@ -762,3 +765,98 @@ func TestAPIDBStalledClientIsCut(t *testing.T) {
<-sr.opened
fx.waitIdle(t, 15*time.Second)
}
+
+// recConn records, in order, every write and every write-deadline change on
+// a server-side connection.
+type recConn struct {
+ net.Conn
+ mu *sync.Mutex
+ events *[]string
+}
+
+func (c recConn) Write(p []byte) (int, error) {
+ c.mu.Lock()
+ *c.events = append(*c.events, "write")
+ c.mu.Unlock()
+ return c.Conn.Write(p)
+}
+
+func (c recConn) SetWriteDeadline(t time.Time) error {
+ c.mu.Lock()
+ if t.IsZero() {
+ *c.events = append(*c.events, "clear")
+ } else {
+ *c.events = append(*c.events, "deadline")
+ }
+ c.mu.Unlock()
+ return c.Conn.SetWriteDeadline(t)
+}
+
+type recListener struct {
+ net.Listener
+ mu sync.Mutex
+ events []string
+}
+
+func (l *recListener) Accept() (net.Conn, error) {
+ c, err := l.Listener.Accept()
+ if err != nil {
+ return nil, err
+ }
+ return recConn{Conn: c, mu: &l.mu, events: &l.events}, nil
+}
+
+// net/http flushes what is still buffered (the last bytes, the chunked
+// terminator) in finishRequest, AFTER the handler returns, and clears the
+// write deadline itself only after that. A handler that clears it on the
+// way out leaves that final flush with no deadline, so a client that stops
+// reading near the end holds the connection forever, outside the stream
+// bound. No byte may ever be written with the deadline cleared.
+func TestAPIDBNeverWritesWithoutADeadline(t *testing.T) {
+ fx := dbFixture(t)
+ fx.replicate(t, dbTestHost, 1)
+ f := listFiles(t, lsfile.NewReplicaClient(filepath.Join(fx.root, "instances", dbTestHost)), 0)[0]
+ ts := httptest.NewUnstartedServer(fx.srv)
+ rl := &recListener{Listener: ts.Listener}
+ ts.Listener = rl
+ ts.Start()
+ defer ts.Close()
+ req, _ := http.NewRequest(http.MethodGet, ts.URL+dbdoor.GetPath("bab", "titogo", dbTestHost, 0, uint64(f.MinTXID), uint64(f.MaxTXID)), nil)
+ req.Header.Set("Authorization", "Bearer "+fx.member)
+ req.Close = true
+ res, err := http.DefaultClient.Do(req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ io.Copy(io.Discard, res.Body)
+ res.Body.Close()
+ if res.StatusCode != http.StatusOK {
+ t.Fatalf("get = %d", res.StatusCode)
+ }
+ deadline := time.Now().Add(5 * time.Second)
+ for {
+ rl.mu.Lock()
+ events := append([]string(nil), rl.events...)
+ rl.mu.Unlock()
+ if len(events) > 0 && events[len(events)-1] == "clear" || time.Now().After(deadline) {
+ armed := false
+ for i, e := range events {
+ switch e {
+ case "deadline":
+ armed = true
+ case "clear":
+ armed = false
+ case "write":
+ if !armed {
+ t.Fatalf("write %d happened with no write deadline: %v", i, events)
+ }
+ }
+ }
+ if !armed && events[len(events)-1] != "clear" {
+ t.Fatalf("events = %v", events)
+ }
+ return
+ }
+ time.Sleep(10 * time.Millisecond)
+ }
+}