From 9ac67cb633cea1bee7a75885f16e62895f80dc91 Mon Sep 17 00:00:00 2001 From: go-while Date: Wed, 11 Nov 2026 13:37:42 +0000 Subject: [PATCH 1/5] e2e: TestClientStalledReader measures the session's end from the spool Old: client A asked for OVER/XOVER 1-40000 (served from the fake upstream, which runs in the race-instrumented test process) and stopped reading; after 2 s client B's GROUP and ARTICLE 39999 had to be answered within 20 s. Nothing checked when A's session ended. Since #274 writes the records with the group's lock released, B is served even when A's session stays blocked in send() for good, so the test no longer saw what 3cefac9 (ClientTimeout for each write) and 763126c (drop the buffer after a failed send) fixed. New, as #369 did for the TLS test: a first client fills the cache, so A's reply comes from the spool and its socket fills at once; A's session pid is the daemon's only child, and a goroutine records when it is gone, measured from just before A's request (it reads /proc//task//children itself, no t.Fatal from the goroutine, and stops with the test). The session must be gone within 1.5 ClientTimeouts (4.5 s); in a file-log build "can't be written to, session ended" must be logged. B's part is unchanged. Needs /proc or the log file, like the TLS test. Measured (12 cores, ClientTimeout 3): the session ended 3.04-3.10 s after the request (5 runs x OVER/XOVER), 3.27-3.37 s with -race; all pass. Mutants (scratch only): 1. 763126c reverted (sockstream.cc: the buffer kept after a failed send): the new test fails, OVER and XOVER: stalled reader's session 2064767 ended 6.035s after the request, want within 4.5s 2. no SO_SNDTIMEO on the client's output (set_socket_timeouts, the core of 3cefac9): the new test fails: stalled reader's session 2065805 still running 14.501s after the request, want it gone within 4.5s The test from origin/main passes against both mutants. Co-Authored-By: Claude Opus 5.5 --- changes/fix-stalled-reader-test-from-spool.md | 10 ++ tests/e2e/daemon_test.go | 107 +++++++++++++++++- 2 files changed, 116 insertions(+), 1 deletion(-) create mode 100644 changes/fix-stalled-reader-test-from-spool.md diff --git a/changes/fix-stalled-reader-test-from-spool.md b/changes/fix-stalled-reader-test-from-spool.md new file mode 100644 index 0000000..13392f2 --- /dev/null +++ b/changes/fix-stalled-reader-test-from-spool.md @@ -0,0 +1,10 @@ +Other changes: +* TestClientStalledReader checks again that the session of a client + that stops reading in the middle of OVER or XOVER ends after one + ClientTimeout: it now fills the cache first, so that the reply comes + from the spool, and wants the session gone within 1.5 ClientTimeouts + of the request (3.0 to 3.3 s, also under -race). It only checked that + another client was still served, which since the records are written + with the group's lock released passed even when the session never + ended (3cefac9 reverted) or ended a ClientTimeout late (763126c + reverted: 6.0 s). diff --git a/tests/e2e/daemon_test.go b/tests/e2e/daemon_test.go index 5cc8ad4..1102363 100644 --- a/tests/e2e/daemon_test.go +++ b/tests/e2e/daemon_test.go @@ -14,6 +14,7 @@ import ( "path/filepath" "regexp" "runtime" + "slices" "strconv" "strings" "sync" @@ -1066,7 +1067,45 @@ func testClientStalledReader(t *testing.T, command string) { e := newEnv(t) e.ClientTimeout = 3 e.Store.AddGenerated(group, "", fakeserver.Generator{Low: 1, High: 40000}) - e.standalone() + mode, err := logMode(e.Bin) + if err != nil { + t.Fatal(err) + } + haveLog := mode != "syslog" + // /proc//task//children (Linux 3.5+); a restricted /proc + // in a container may lack it + self := strconv.Itoa(os.Getpid()) + _, procErr := os.ReadFile(filepath.Join("/proc", self, "task", self, "children")) + haveProc := runtime.GOOS == "linux" && procErr == nil + if !haveLog && !haveProc { + t.Skip("needs the log file (configure --without-syslog) or /proc (Linux) to see the session end") + } + p := e.standalone() + if haveProc { + // standalone()'s readiness check is a session of its own + waitForNoLinuxChildren(t, p.Pid(), 10*time.Second) + } + + // Fill the cache first, so that A's reply below comes from the spool + // and its socket fills at once: the time to the session's end must not + // count how fast the fake upstream, which runs in this process (slow + // under -race), serves 40000 records. + w := e.dial() + w.Deadline = 60 * time.Second + if code, line := w.Cmd("GROUP %s", group); code != 211 { + t.Fatalf("warming GROUP: %q", line) + } + if code, line := w.Cmd("%s 1-40000", command); code != 224 { + t.Fatalf("warming %s: %q", command, line) + } + if n := len(w.ReadDotLines()); n != 40000 { + t.Fatalf("warming %s returned %d records, want 40000", command, n) + } + w.Cmd("QUIT") + if haveProc { + waitForNoLinuxChildren(t, p.Pid(), 10*time.Second) + } + a, err := net.Dial("tcp", e.Listen) if err != nil { t.Fatal(err) @@ -1080,6 +1119,40 @@ func testClientStalledReader(t *testing.T, command string) { if l, err := ra.ReadString('\n'); err != nil || !strings.HasPrefix(l, "211") { t.Fatalf("A: GROUP: %q %v", l, err) } + // The daemon's only child is A's session. + var aPid string + if haveProc { + kids := linuxProcChildren(t, p.Pid()) + if len(kids) != 1 { + t.Fatalf("daemon children before the stall = %v, want A's session", kids) + } + aPid = kids[0] + } + stallStart := time.Now() + // When A's session ends, seen from the request: checked at the end, + // after B, which a busy machine slows down. + // (Not through linuxProcChildren: no t.Fatal from this goroutine, + // and it stops with the test when the session doesn't end.) + sessionEnd := make(chan time.Duration, 1) + stop := make(chan struct{}) + defer close(stop) + if haveProc { + children := filepath.Join("/proc", strconv.Itoa(p.Pid()), "task", strconv.Itoa(p.Pid()), "children") + go func() { + for { + b, err := os.ReadFile(children) + if err == nil && !slices.Contains(strings.Fields(string(b)), aPid) { + sessionEnd <- time.Since(stallStart) + return + } + select { + case <-stop: + return + case <-time.After(20 * time.Millisecond): + } + } + }() + } io.WriteString(a, command+" 1-40000\r\n") // A reads the status line and a little, then nothing more if l, err := ra.ReadString('\n'); err != nil || !strings.HasPrefix(l, "224") { @@ -1098,6 +1171,38 @@ func testClientStalledReader(t *testing.T, command string) { } b.ReadDotLines() t.Logf("B served after %v", time.Since(start).Round(100*time.Millisecond)) + + // A's session must end one ClientTimeout after the request (the send + // timeout): 3.0 to 3.1 s with ClientTimeout 3, 3.3 s under -race. It + // used to stay blocked in send() for good (3cefac9), and a failed send + // that left its data in the buffer made the next flush wait for the + // timeout again: 6.0 s here (763126c). The bound is 1.5 ClientTimeouts. + endBound := time.Duration(e.ClientTimeout) * 1500 * time.Millisecond + if haveProc { + select { + case d := <-sessionEnd: + t.Logf("the stalled session ended %v after the request", d.Round(time.Millisecond)) + if d > endBound { + t.Errorf("stalled reader's session %s ended %v after the request, want within %v", aPid, d.Round(time.Millisecond), endBound) + } + case <-time.After(endBound - time.Since(stallStart) + 10*time.Second): + t.Errorf("stalled reader's session %s still running %v after the request, want it gone within %v", aPid, time.Since(stallStart).Round(time.Millisecond), endBound) + } + } + if !haveLog { + return + } + for { + log := e.logFile() + if strings.Contains(log, "can't be written to, session ended") { + break + } + if time.Since(stallStart) > endBound+10*time.Second { + t.Errorf("stalled reader's session not logged as ended %v after the request", time.Since(stallStart).Round(time.Millisecond)) + break + } + time.Sleep(50 * time.Millisecond) + } } // TestClientPostStall: a client that sends POST, part of the article and From 930073c4e70d5d36f03106ef4d21ca874b0519d1 Mon Sep 17 00:00:00 2001 From: go-while Date: Wed, 11 Nov 2026 18:45:25 +0000 Subject: [PATCH 2/5] e2e: TestClientStalledReader sees the session end without /proc too Without /proc the test fell back to the log line "can't be written to, session ended", which is logged at the first failed write, not when the session ends: with 763126c reverted it comes 3 s before the end, so the bound wasn't checked there (Terra's finding). The session end now always comes from the daemon's children, through daemonChildren (/proc, or pgrep -P, from #369); the log line stays as a check of why the session ended. The test skips only where neither /proc nor pgrep exists. With /proc hidden (scratch change): main 3.10-3.14 s, pass; 763126c reverted 6.14-6.16 s, fails. Co-Authored-By: Claude Opus 5.5 --- tests/e2e/daemon_test.go | 79 +++++++++++++++++----------------------- 1 file changed, 33 insertions(+), 46 deletions(-) diff --git a/tests/e2e/daemon_test.go b/tests/e2e/daemon_test.go index d6099b9..aaa8bb4 100644 --- a/tests/e2e/daemon_test.go +++ b/tests/e2e/daemon_test.go @@ -1078,19 +1078,14 @@ func testClientStalledReader(t *testing.T, command string) { t.Fatal(err) } haveLog := mode != "syslog" - // /proc//task//children (Linux 3.5+); a restricted /proc - // in a container may lack it - self := strconv.Itoa(os.Getpid()) - _, procErr := os.ReadFile(filepath.Join("/proc", self, "task", self, "children")) - haveProc := runtime.GOOS == "linux" && procErr == nil - if !haveLog && !haveProc { - t.Skip("needs the log file (configure --without-syslog) or /proc (Linux) to see the session end") + // A's session end is seen in the daemon's children: /proc, or pgrep -P + // without /proc (FreeBSD, macOS) + if _, ok := daemonChildren(os.Getpid()); !ok { + t.Skip("can't list a process's children (no /proc, no pgrep) to see the session end") } p := e.standalone() - if haveProc { - // standalone()'s readiness check is a session of its own - waitForNoLinuxChildren(t, p.Pid(), 10*time.Second) - } + // standalone()'s readiness check is a session of its own + waitForNoLinuxChildren(t, p.Pid(), 10*time.Second) // Fill the cache first, so that A's reply below comes from the spool // and its socket fills at once: the time to the session's end must not @@ -1108,9 +1103,7 @@ func testClientStalledReader(t *testing.T, command string) { t.Fatalf("warming %s returned %d records, want 40000", command, n) } w.Cmd("QUIT") - if haveProc { - waitForNoLinuxChildren(t, p.Pid(), 10*time.Second) - } + waitForNoLinuxChildren(t, p.Pid(), 10*time.Second) a, err := net.Dial("tcp", e.Listen) if err != nil { @@ -1126,14 +1119,11 @@ func testClientStalledReader(t *testing.T, command string) { t.Fatalf("A: GROUP: %q %v", l, err) } // The daemon's only child is A's session. - var aPid string - if haveProc { - kids := linuxProcChildren(t, p.Pid()) - if len(kids) != 1 { - t.Fatalf("daemon children before the stall = %v, want A's session", kids) - } - aPid = kids[0] + kids := linuxProcChildren(t, p.Pid()) + if len(kids) != 1 { + t.Fatalf("daemon children before the stall = %v, want A's session", kids) } + aPid := kids[0] stallStart := time.Now() // When A's session ends, seen from the request: checked at the end, // after B, which a busy machine slows down. @@ -1142,23 +1132,20 @@ func testClientStalledReader(t *testing.T, command string) { sessionEnd := make(chan time.Duration, 1) stop := make(chan struct{}) defer close(stop) - if haveProc { - children := filepath.Join("/proc", strconv.Itoa(p.Pid()), "task", strconv.Itoa(p.Pid()), "children") - go func() { - for { - b, err := os.ReadFile(children) - if err == nil && !slices.Contains(strings.Fields(string(b)), aPid) { - sessionEnd <- time.Since(stallStart) - return - } - select { - case <-stop: - return - case <-time.After(20 * time.Millisecond): - } + go func() { + for { + kids, ok := daemonChildren(p.Pid()) + if ok && !slices.Contains(kids, aPid) { + sessionEnd <- time.Since(stallStart) + return } - }() - } + select { + case <-stop: + return + case <-time.After(20 * time.Millisecond): + } + } + }() io.WriteString(a, command+" 1-40000\r\n") // A reads the status line and a little, then nothing more if l, err := ra.ReadString('\n'); err != nil || !strings.HasPrefix(l, "224") { @@ -1184,17 +1171,17 @@ func testClientStalledReader(t *testing.T, command string) { // that left its data in the buffer made the next flush wait for the // timeout again: 6.0 s here (763126c). The bound is 1.5 ClientTimeouts. endBound := time.Duration(e.ClientTimeout) * 1500 * time.Millisecond - if haveProc { - select { - case d := <-sessionEnd: - t.Logf("the stalled session ended %v after the request", d.Round(time.Millisecond)) - if d > endBound { - t.Errorf("stalled reader's session %s ended %v after the request, want within %v", aPid, d.Round(time.Millisecond), endBound) - } - case <-time.After(endBound - time.Since(stallStart) + 10*time.Second): - t.Errorf("stalled reader's session %s still running %v after the request, want it gone within %v", aPid, time.Since(stallStart).Round(time.Millisecond), endBound) + select { + case d := <-sessionEnd: + t.Logf("the stalled session ended %v after the request", d.Round(time.Millisecond)) + if d > endBound { + t.Errorf("stalled reader's session %s ended %v after the request, want within %v", aPid, d.Round(time.Millisecond), endBound) } + case <-time.After(endBound - time.Since(stallStart) + 10*time.Second): + t.Errorf("stalled reader's session %s still running %v after the request, want it gone within %v", aPid, time.Since(stallStart).Round(time.Millisecond), endBound) } + // With a log file: the session ended because the send failed. (Not + // the time of its end: that line comes at the first failed write.) if !haveLog { return } From 21e5d48225e16ac32cffbbe3413235107083ccf9 Mon Sep 17 00:00:00 2001 From: go-while Date: Wed, 11 Nov 2026 19:21:40 +0000 Subject: [PATCH 3/5] Plain sockets: a send cut short by the send timeout ends the write When SO_SNDTIMEO runs out after send() has taken part of the data, FreeBSD returns that part instead of EAGAIN (kern_sendit turns EWOULDBLOCK into a short count). sync() took the short count for progress and sent the rest, which waited for a second timeout: the session of a client that stopped reading ended after two ClientTimeouts. FreeBSD CI showed it with the new TestClientStalledReader: 6.088 s (OVER) and 6.044 s (XOVER) with ClientTimeout 3, bound 4.5 s. A blocking send returns short only when it was cut off, by a signal or by the timeout. When it has waited for the socket's whole send timeout (read with getsockopt, since a client's is set on its socket, not through set_io_timeout), the rest is dropped and the write fails, as after a failed send. A short count after a shorter wait (a signal) still goes on. Co-Authored-By: Claude Opus 5.5 --- NServer-0.9.1/sockstream.cc | 35 ++++++++++++++++++- NServer-0.9.1/sockstream.h | 1 + changes/fix-stalled-reader-test-from-spool.md | 8 +++++ 3 files changed, 43 insertions(+), 1 deletion(-) diff --git a/NServer-0.9.1/sockstream.cc b/NServer-0.9.1/sockstream.cc index a8be565..a5bf706 100644 --- a/NServer-0.9.1/sockstream.cc +++ b/NServer-0.9.1/sockstream.cc @@ -568,6 +568,24 @@ int sockstream::sockstreambuf::overflow(int c) return c; } +// Whether a send that began at t0 has waited for the socket's whole send +// timeout (SO_SNDTIMEO: ClientTimeout for a client, set on its socket, or +// the I/O timeout). 90 %: the kernel counts the timeout in clock ticks. +bool sockstream::sockstreambuf::send_timed_out(const struct timespec &t0) const +{ + struct timeval tv; + socklen_t len = sizeof(tv); + struct timespec t1; + if (getsockopt(_fd, SOL_SOCKET, SO_SNDTIMEO, &tv, &len) != 0 || + (tv.tv_sec == 0 && tv.tv_usec == 0) || + clock_gettime(CLOCK_MONOTONIC, &t1) != 0) + return false; + const long long waited = ((long long) t1.tv_sec - t0.tv_sec) * 1000 + + ((long long) t1.tv_nsec - t0.tv_nsec) / 1000000; + const long long limit = (long long) tv.tv_sec * 1000 + tv.tv_usec / 1000; + return waited * 10 >= limit * 9; +} + int sockstream::sockstreambuf::sync() { size_t off = 0; @@ -582,9 +600,12 @@ int sockstream::sockstreambuf::sync() { return -1; } + const size_t want = pptr() - pbase() - off; + struct timespec t0; + const bool timed = clock_gettime(CLOCK_MONOTONIC, &t0) == 0; do { - rc_send = ::send(_fd, pbase() + off, pptr() - pbase() - off, 0); + rc_send = ::send(_fd, pbase() + off, want, 0); } while ((rc_send < 0) && (errno == EINTR) && !interrupted()); @@ -598,6 +619,18 @@ int sockstream::sockstreambuf::sync() } off += rc_send; + + // A blocking send returns less than asked only when it was cut + // short: by a signal, or by the send timeout (SO_SNDTIMEO) after it + // had taken part of the data, which FreeBSD returns as a short count + // instead of EAGAIN. After a whole timeout the client isn't reading: + // that is final too. Sending the rest waited for a second timeout, + // and a session stalled for two ClientTimeouts instead of one. + if ((size_t) rc_send < want && timed && send_timed_out(t0)) + { + setp(_pbuf, _pbuf + sizeof(_pbuf)); + return -1; + } } setp(_pbuf, _pbuf + sizeof(_pbuf)); diff --git a/NServer-0.9.1/sockstream.h b/NServer-0.9.1/sockstream.h index e1da7bd..db3e89f 100644 --- a/NServer-0.9.1/sockstream.h +++ b/NServer-0.9.1/sockstream.h @@ -109,6 +109,7 @@ class sockstream char _pbuf[4096]; void apply_io_timeout(); + bool send_timed_out(const struct timespec &t0) const; // seconds wait_io() waits for reading / writing, 0 = _io_timeout unsigned int _wait_read, _wait_write; diff --git a/changes/fix-stalled-reader-test-from-spool.md b/changes/fix-stalled-reader-test-from-spool.md index 13392f2..29fe7ce 100644 --- a/changes/fix-stalled-reader-test-from-spool.md +++ b/changes/fix-stalled-reader-test-from-spool.md @@ -1,3 +1,11 @@ +Bug fixes: +* On FreeBSD the session of a client that stopped reading in the middle + of a reply ended after two ClientTimeouts instead of one: when the send + timeout runs out after send() has taken part of the data, FreeBSD + returns that part instead of an error, and the rest was sent with a + second wait. A send that returns short after waiting for the whole + send timeout now ends the write, like a failed one. + Other changes: * TestClientStalledReader checks again that the session of a client that stops reading in the middle of OVER or XOVER ends after one From fa0d23f937fb94401218d41f2ef061d1a782a068 Mon Sep 17 00:00:00 2001 From: go-while Date: Wed, 11 Nov 2026 19:36:39 +0000 Subject: [PATCH 4/5] Plain sockets: after a short send, wait only the rest of the send timeout Terra's review of 21e5d48: the 90 % check guessed from the elapsed time whether the timeout or a signal cut the send short, and t0 was taken before the EINTR retries. Now a short send waits until the socket can take more for what is left of the send timeout (poll), from the start of the send call that returned short: nothing left or still full ends the write, so a signal never ends it early and the timeout never runs twice. Not during a reply deadline (non-blocking socket, short counts at once): that path is unchanged. Linux returns short counts at the timeout too (1774 of 4096 bytes after 1 s, then EAGAIN after another second), so cReplyBudget gets a check: loopback TCP with a small window, SO_SNDTIMEO 1 s, a peer that reads nothing; the write must fail within 1.5 s. main's sockstream: 2.06-2.07 s, fails 3/3; with this: 1.00-1.06 s. Co-Authored-By: Claude Opus 5.5 --- NServer-0.9.1/checks/cReplyBudget.cc | 57 +++++++++++++++++++ NServer-0.9.1/sockstream.cc | 52 +++++++++++------ NServer-0.9.1/sockstream.h | 2 +- changes/fix-stalled-reader-test-from-spool.md | 14 +++-- 4 files changed, 102 insertions(+), 23 deletions(-) diff --git a/NServer-0.9.1/checks/cReplyBudget.cc b/NServer-0.9.1/checks/cReplyBudget.cc index 455ff34..ca0f544 100644 --- a/NServer-0.9.1/checks/cReplyBudget.cc +++ b/NServer-0.9.1/checks/cReplyBudget.cc @@ -1,9 +1,12 @@ #include #include #include +#include #include #include #include +#include +#include #include #include "Logger.h" #include "ReplyBudget.h" @@ -97,6 +100,59 @@ static void client_connection_gone() close(fd[1]); } +// When the send timeout runs out after send() has taken part of the data, +// Linux and FreeBSD return that part instead of EAGAIN. The rest must not +// wait for a second timeout: the write to a peer that reads nothing fails +// after one (it took two, and a stalled client's session two +// ClientTimeouts). +static void short_send_ends_write() +{ + int l = socket(AF_INET, SOCK_STREAM, 0); + struct sockaddr_in sa; + socklen_t salen = sizeof(sa); + memset(&sa, 0, sizeof(sa)); + sa.sin_family = AF_INET; + sa.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + int small = 4096; + // (a small window, inherited by the accepted socket) + setsockopt(l, SOL_SOCKET, SO_RCVBUF, &small, sizeof(small)); + if (l < 0 || bind(l, (struct sockaddr *) &sa, sizeof(sa)) != 0 || + listen(l, 1) != 0 || + getsockname(l, (struct sockaddr *) &sa, &salen) != 0) { + std::cerr << "short_send_ends_write: no loopback socket, skipped\n"; + if (l >= 0) + close(l); + return; + } + int c = socket(AF_INET, SOCK_STREAM, 0); + CHECK(c >= 0 && connect(c, (struct sockaddr *) &sa, sizeof(sa)) == 0); + int a = accept(l, NULL, NULL); + CHECK(a >= 0); + { + sockstream stream; + stream.attach(c); + stream.set_io_timeout(1); // SO_SNDTIMEO 1 s + const std::string chunk(1000, 'x'); + struct timespec before, after; + clock_gettime(CLOCK_MONOTONIC, &before); + for (int i = 0; i < 1000000 && stream.good(); ++i) + stream << chunk; + stream.flush(); + clock_gettime(CLOCK_MONOTONIC, &after); + double elapsed = after.tv_sec - before.tv_sec + + (after.tv_nsec - before.tv_nsec) / 1000000000.0; + std::cerr << "short_send_ends_write: the write failed after " + << elapsed << " s\n"; + CHECK(!stream.good()); + // one send timeout (1 s), not two + CHECK(elapsed < 1.5); + stream.abort(); + } + if (a >= 0) + close(a); + close(l); +} + int main() { #if defined(NEWSCACHE_TEST_SHORT_REPLY_LIMITS) @@ -154,5 +210,6 @@ int main() CHECK(line == "abc" && wire == 3 && partial.eof()); partial_line_deadline(); client_connection_gone(); + short_send_ends_write(); return failures ? 1 : 0; } diff --git a/NServer-0.9.1/sockstream.cc b/NServer-0.9.1/sockstream.cc index a5bf706..1ce5baa 100644 --- a/NServer-0.9.1/sockstream.cc +++ b/NServer-0.9.1/sockstream.cc @@ -568,22 +568,37 @@ int sockstream::sockstreambuf::overflow(int c) return c; } -// Whether a send that began at t0 has waited for the socket's whole send -// timeout (SO_SNDTIMEO: ClientTimeout for a client, set on its socket, or -// the I/O timeout). 90 %: the kernel counts the timeout in clock ticks. -bool sockstream::sockstreambuf::send_timed_out(const struct timespec &t0) const +// After a send that began at t0 returned short: waits until the socket +// can take more, for what is left of its send timeout (SO_SNDTIMEO: +// ClientTimeout for a client, set on its socket, or the I/O timeout). +// False when nothing is left or the socket stays full: the client isn't +// reading. Without a send timeout, true (the next send waits as before). +bool sockstream::sockstreambuf::wait_rest_of_send_timeout(const struct timespec &t0) { struct timeval tv; socklen_t len = sizeof(tv); - struct timespec t1; if (getsockopt(_fd, SOL_SOCKET, SO_SNDTIMEO, &tv, &len) != 0 || - (tv.tv_sec == 0 && tv.tv_usec == 0) || - clock_gettime(CLOCK_MONOTONIC, &t1) != 0) - return false; - const long long waited = ((long long) t1.tv_sec - t0.tv_sec) * 1000 + - ((long long) t1.tv_nsec - t0.tv_nsec) / 1000000; + (tv.tv_sec == 0 && tv.tv_usec == 0)) + return true; const long long limit = (long long) tv.tv_sec * 1000 + tv.tv_usec / 1000; - return waited * 10 >= limit * 9; + int rc; + do + { + struct timespec t1; + if (interrupted() || clock_gettime(CLOCK_MONOTONIC, &t1) != 0) + return false; + const long long left = limit - + (((long long) t1.tv_sec - t0.tv_sec) * 1000 + + ((long long) t1.tv_nsec - t0.tv_nsec) / 1000000); + if (left <= 0) + return false; + struct pollfd p; + p.fd = _fd; + p.events = POLLOUT; + p.revents = 0; + rc = poll(&p, 1, left > INT_MAX ? INT_MAX : (int) left); + } while (rc < 0 && errno == EINTR); + return rc > 0; } int sockstream::sockstreambuf::sync() @@ -602,9 +617,10 @@ int sockstream::sockstreambuf::sync() } const size_t want = pptr() - pbase() - off; struct timespec t0; - const bool timed = clock_gettime(CLOCK_MONOTONIC, &t0) == 0; + bool timed; do { + timed = clock_gettime(CLOCK_MONOTONIC, &t0) == 0; rc_send = ::send(_fd, pbase() + off, want, 0); } while ((rc_send < 0) && (errno == EINTR) && !interrupted()); @@ -623,10 +639,14 @@ int sockstream::sockstreambuf::sync() // A blocking send returns less than asked only when it was cut // short: by a signal, or by the send timeout (SO_SNDTIMEO) after it // had taken part of the data, which FreeBSD returns as a short count - // instead of EAGAIN. After a whole timeout the client isn't reading: - // that is final too. Sending the rest waited for a second timeout, - // and a session stalled for two ClientTimeouts instead of one. - if ((size_t) rc_send < want && timed && send_timed_out(t0)) + // instead of EAGAIN. The rest is sent only if the socket can take + // more within what is left of the timeout; after the whole timeout + // the client isn't reading, and that is final too. Sending the rest + // at once waited for a second timeout: a session stalled for two + // ClientTimeouts instead of one. (Not during a reply deadline: the + // socket is non-blocking then, and a short count comes at once.) + if ((size_t) rc_send < want && timed && !_reply_deadline_active && + !wait_rest_of_send_timeout(t0)) { setp(_pbuf, _pbuf + sizeof(_pbuf)); return -1; diff --git a/NServer-0.9.1/sockstream.h b/NServer-0.9.1/sockstream.h index db3e89f..6290f62 100644 --- a/NServer-0.9.1/sockstream.h +++ b/NServer-0.9.1/sockstream.h @@ -109,7 +109,7 @@ class sockstream char _pbuf[4096]; void apply_io_timeout(); - bool send_timed_out(const struct timespec &t0) const; + bool wait_rest_of_send_timeout(const struct timespec &t0); // seconds wait_io() waits for reading / writing, 0 = _io_timeout unsigned int _wait_read, _wait_write; diff --git a/changes/fix-stalled-reader-test-from-spool.md b/changes/fix-stalled-reader-test-from-spool.md index 29fe7ce..eaf6ff7 100644 --- a/changes/fix-stalled-reader-test-from-spool.md +++ b/changes/fix-stalled-reader-test-from-spool.md @@ -1,10 +1,12 @@ Bug fixes: -* On FreeBSD the session of a client that stopped reading in the middle - of a reply ended after two ClientTimeouts instead of one: when the send - timeout runs out after send() has taken part of the data, FreeBSD - returns that part instead of an error, and the rest was sent with a - second wait. A send that returns short after waiting for the whole - send timeout now ends the write, like a failed one. +* The session of a client that stopped reading in the middle of a reply + could end after two ClientTimeouts instead of one (seen on FreeBSD): + when the send timeout runs out after send() has taken part of the + data, Linux and FreeBSD return that part instead of an error, and the + rest was sent with a second wait. After a short send, the rest now + goes out only if the socket can take more within what is left of the + send timeout; otherwise the write fails, like a failed send. The same + applies to connections to the upstream servers (ReadTimeout). Other changes: * TestClientStalledReader checks again that the session of a client From 1af47d50f2ec7cc953d58af2845b901272dfb6d7 Mon Sep 17 00:00:00 2001 From: go-while Date: Wed, 11 Nov 2026 19:41:42 +0000 Subject: [PATCH 5/5] cReplyBudget: the short-send check returns when it can't connect On a failed socket() or connect() it went on to accept(), which waits for good with no peer coming (Terra's review). Now it reports the failure and returns; a failed accept() too. Co-Authored-By: Claude Opus 5.5 --- NServer-0.9.1/checks/cReplyBudget.cc | 19 +++++++++++++++---- 1 file changed, 15 insertions(+), 4 deletions(-) diff --git a/NServer-0.9.1/checks/cReplyBudget.cc b/NServer-0.9.1/checks/cReplyBudget.cc index ca0f544..577f0c2 100644 --- a/NServer-0.9.1/checks/cReplyBudget.cc +++ b/NServer-0.9.1/checks/cReplyBudget.cc @@ -124,10 +124,22 @@ static void short_send_ends_write() close(l); return; } + // (on a failure no accept(): it would wait for good) int c = socket(AF_INET, SOCK_STREAM, 0); - CHECK(c >= 0 && connect(c, (struct sockaddr *) &sa, sizeof(sa)) == 0); + if (c < 0 || connect(c, (struct sockaddr *) &sa, sizeof(sa)) != 0) { + CHECK(!"connect to the loopback listener"); + if (c >= 0) + close(c); + close(l); + return; + } int a = accept(l, NULL, NULL); - CHECK(a >= 0); + if (a < 0) { + CHECK(!"accept"); + close(c); + close(l); + return; + } { sockstream stream; stream.attach(c); @@ -148,8 +160,7 @@ static void short_send_ends_write() CHECK(elapsed < 1.5); stream.abort(); } - if (a >= 0) - close(a); + close(a); close(l); }