Enforce client output buffer soft limit when no traffic. (#8833)

When client breached the output buffer soft limit but then went idle,
we didn't disconnect on soft limit timeout, now we do.
Note this also resolves some sporadic test failures in due to Linux
buffering data which caused tests to fail if during the test we went
back under the soft COB limit.

Co-authored-by: Oran Agra <oran@redislabs.com>
Co-authored-by: sundb <sundbcn@gmail.com>
(cherry picked from commit 152fce5e2cbf947a389da414a431f7331981a374)

src/networking.c

@@ -333,7 +333,7 @@ void _addReplyProtoToList(client *c, const char *s, size_t len) {
         listAddNodeTail(c->reply, tail);
         c->reply_bytes += tail->size;
 
-        asyncCloseClientOnOutputBufferLimitReached(c);
+        closeClientOnOutputBufferLimitReached(c, 1);
     }
 }
 
@@ -616,7 +616,7 @@ void setDeferredReply(client *c, void *node, const char *s, size_t length) {
         listNodeValue(ln) = buf;
         c->reply_bytes += buf->size;
 
-        asyncCloseClientOnOutputBufferLimitReached(c);
+        closeClientOnOutputBufferLimitReached(c, 1);
     }
 }
 
@@ -949,7 +949,7 @@ void AddReplyFromClient(client *dst, client *src) {
     src->bufpos = 0;
 
     /* Check output buffer limits */
-    asyncCloseClientOnOutputBufferLimitReached(dst);
+    closeClientOnOutputBufferLimitReached(dst, 1);
 }
 
 /* Copy 'src' client output buffers into 'dst' client output buffers.
@@ -3223,18 +3223,33 @@ int checkClientOutputBufferLimits(client *c) {
  *
  * Note: we need to close the client asynchronously because this function is
  * called from contexts where the client can't be freed safely, i.e. from the
- * lower level functions pushing data inside the client output buffers. */
+ * lower level functions pushing data inside the client output buffers.
-void asyncCloseClientOnOutputBufferLimitReached(client *c) {
+ * When `async` is set to 0, we close the client immediately, this is
-    if (!c->conn) return; /* It is unsafe to free fake clients. */
+ * useful when called from cron.
+ *
+ * Returns 1 if client was (flagged) closed. */
+int closeClientOnOutputBufferLimitReached(client *c, int async) {
+    if (!c->conn) return 0; /* It is unsafe to free fake clients. */
     serverAssert(c->reply_bytes < SIZE_MAX-(1024*64));
-    if (c->reply_bytes == 0 || c->flags & CLIENT_CLOSE_ASAP) return;
+    if (c->reply_bytes == 0 || c->flags & CLIENT_CLOSE_ASAP) return 0;
     if (checkClientOutputBufferLimits(c)) {
         sds client = catClientInfoString(sdsempty(),c);
 
-        freeClientAsync(c);
+        if (async) {
-        serverLog(LL_WARNING,"Client %s scheduled to be closed ASAP for overcoming of output buffer limits.", client);
+            freeClientAsync(c);
+            serverLog(LL_WARNING,
+                      "Client %s scheduled to be closed ASAP for overcoming of output buffer limits.",
+                      client);
+        } else {
+            freeClient(c);
+            serverLog(LL_WARNING,
+                      "Client %s closed for overcoming of output buffer limits.",
+                      client);
+        }
         sdsfree(client);
+        return  1;
     }
+    return 0;
 }
 
 /* Helper function used by performEvictions() in order to flush slaves

src/server.c

@@ -1839,6 +1839,7 @@ void clientsCron(void) {
         if (clientsCronResizeQueryBuffer(c)) continue;
         if (clientsCronTrackExpansiveClients(c, curr_peak_mem_usage_slot)) continue;
         if (clientsCronTrackClientsMemUsage(c)) continue;
+        if (closeClientOnOutputBufferLimitReached(c, 0)) continue;
     }
 }
 

src/server.h

@@ -1867,7 +1867,7 @@ void rewriteClientCommandArgument(client *c, int i, robj *newval);
 void replaceClientCommandVector(client *c, int argc, robj **argv);
 unsigned long getClientOutputBufferMemoryUsage(client *c);
 int freeClientsInAsyncFreeQueue(void);
-void asyncCloseClientOnOutputBufferLimitReached(client *c);
+int closeClientOnOutputBufferLimitReached(client *c, int async);
 int getClientType(client *c);
 int getClientTypeByName(char *name);
 char *getClientTypeName(int class);

tests/unit/obuf-limits.tcl

@@ -18,65 +18,72 @@ start_server {tags {"obuf-limits"}} {
         assert {$omem >= 70000 && $omem < 200000}
         $rd1 close
     }
-
+    
-    test {Client output buffer soft limit is not enforced if time is not overreached} {
+    foreach {soft_limit_time wait_for_timeout} {3 yes
-        r config set client-output-buffer-limit {pubsub 0 100000 10}
+                                                4 no } {
-        set rd1 [redis_deferring_client]
+        if $wait_for_timeout {
-
+            set test_name "Client output buffer soft limit is enforced if time is overreached"
-        $rd1 subscribe foo
+        } else {
-        set reply [$rd1 read]
+            set test_name "Client output buffer soft limit is not enforced too early and is enforced when no traffic"
-        assert {$reply eq "subscribe foo 1"}
-
-        set omem 0
-        set start_time 0
-        set time_elapsed 0
-        while 1 {
-            if {$start_time != 0} {
-                # Slow down loop when omen has reached the limit.
-                after 10
-            }
-            r publish foo [string repeat "x" 1000]
-            set clients [split [r client list] "\r\n"]
-            set c [split [lindex $clients 1] " "]
-            if {![regexp {omem=([0-9]+)} $c - omem]} break
-            if {$omem > 100000} {
-                if {$start_time == 0} {set start_time [clock seconds]}
-                set time_elapsed [expr {[clock seconds]-$start_time}]
-                if {$time_elapsed >= 5} break
-            }
         }
-        assert {$omem >= 100000 && $time_elapsed >= 5 && $time_elapsed <= 10}
-        $rd1 close
-    }
 
-    test {Client output buffer soft limit is enforced if time is overreached} {
+        test $test_name {
-        r config set client-output-buffer-limit {pubsub 0 100000 3}
+            r config set client-output-buffer-limit "pubsub 0 100000 $soft_limit_time"
-        set rd1 [redis_deferring_client]
+            set soft_limit_time [expr $soft_limit_time*1000]
+            set rd1 [redis_deferring_client]
 
-        $rd1 subscribe foo
+            $rd1 client setname test_client
-        set reply [$rd1 read]
+            set reply [$rd1 read]
-        assert {$reply eq "subscribe foo 1"}
+            assert {$reply eq "OK"}
 
-        set omem 0
+            $rd1 subscribe foo
-        set start_time 0
+            set reply [$rd1 read]
-        set time_elapsed 0
+            assert {$reply eq "subscribe foo 1"}
-        while 1 {
+
-            if {$start_time != 0} {
+            set omem 0
-                # Slow down loop when omen has reached the limit.
+            set start_time 0
-                after 10
+            set time_elapsed 0
+            set last_under_limit_time [clock milliseconds]
+            while 1 {
+                r publish foo [string repeat "x" 1000]
+                set clients [split [r client list] "\r\n"]
+                set c [lsearch -inline $clients *name=test_client*]
+                if {$start_time != 0} {
+                    set time_elapsed [expr {[clock milliseconds]-$start_time}]
+                    # Make sure test isn't taking too long
+                    assert {$time_elapsed <= [expr $soft_limit_time+3000]}
+                }
+                if {$wait_for_timeout && $c == ""} {
+                    # Make sure we're disconnected when we reach the soft limit
+                    assert {$omem >= 100000 && $time_elapsed >= $soft_limit_time}
+                    break
+                } else {
+                    assert {[regexp {omem=([0-9]+)} $c - omem]}
+                }
+                if {$omem > 100000} {
+                    if {$start_time == 0} {set start_time $last_under_limit_time}
+                    if {!$wait_for_timeout && $time_elapsed >= [expr $soft_limit_time-1000]} break
+                    # Slow down loop when omem has reached the limit.
+                    after 10
+                } else {
+                    # if the OS socket buffers swallowed what we previously filled, reset the start timer.
+                    set start_time 0
+                    set last_under_limit_time [clock milliseconds]
+                }
             }
-            r publish foo [string repeat "x" 1000]
+
-            set clients [split [r client list] "\r\n"]
+            if {!$wait_for_timeout} {
-            set c [split [lindex $clients 1] " "]
+                # After we completely stopped the traffic, wait for soft limit to time out
-            if {![regexp {omem=([0-9]+)} $c - omem]} break
+                set timeout [expr {$soft_limit_time+1500 - ([clock milliseconds]-$start_time)}]
-            if {$omem > 100000} {
+                wait_for_condition [expr $timeout/10] 10 {
-                if {$start_time == 0} {set start_time [clock seconds]}
+                    [lsearch [split [r client list] "\r\n"] *name=test_client*] == -1
-                set time_elapsed [expr {[clock seconds]-$start_time}]
+                } else {
-                if {$time_elapsed >= 10} break
+                    fail "Soft limit timed out but client still connected"
+                }
             }
+
+            $rd1 close
         }
-        assert {$omem >= 100000 && $time_elapsed < 6}
-        $rd1 close
     }
 
     test {No response for single command if client output buffer hard limit is enforced} {