BUG/MINOR: sink: fix a race condition in the TCP log forwarding code

That's exactly the same as commit 53bfab080c ("BUG/MINOR: sink: fix a race
condition between the writer and the reader") that went into 2.7 and was
backported as far as 2.4, except that since the code was duplicated, the
second instance was not noticed, leaving the race present. The race has
a limited impact, if a forwarder reaches the end of the logs and a new
message arrives before it leaves, the forwarder will only wake up after
yet another new message will be sent. In practice it remains unnoticeable
because for the race to trigger, one needs to have a steady flow of logs,
which means the wakeup will happen anyway.

This should be backported, but no need to insist on it if it resists.

src/sink.c

@@ -440,7 +440,7 @@ static void sink_forward_oc_io_handler(struct appctx *appctx)
 	struct ring *ring = sink->ctx.ring;
 	struct buffer *buf = &ring->buf;
 	uint64_t msg_len;
-	size_t len, cnt, ofs;
+	size_t len, cnt, ofs, last_ofs;
 	int ret = 0;
 	char *p;
 
@@ -531,16 +531,24 @@ static void sink_forward_oc_io_handler(struct appctx *appctx)
 	}
 
 	HA_ATOMIC_INC(b_peek(buf, ofs));
+	last_ofs = b_tail_ofs(buf);
 	sft->ofs = b_peek_ofs(buf, ofs);
-
 	HA_RWLOCK_RDUNLOCK(RING_LOCK, &ring->lock);
 
 	if (ret) {
 		/* let's be woken up once new data arrive */
 		HA_RWLOCK_WRLOCK(RING_LOCK, &ring->lock);
 		LIST_APPEND(&ring->waiters, &appctx->wait_entry);
+		ofs = b_tail_ofs(buf);
 		HA_RWLOCK_WRUNLOCK(RING_LOCK, &ring->lock);
-		applet_have_no_more_data(appctx);
+		if (ofs != last_ofs) {
+			/* more data was added into the ring between the
+			 * unlock and the lock, and the writer might not
+			 * have seen us. We need to reschedule a read.
+			 */
+			applet_have_more_data(appctx);
+		} else
+			applet_have_no_more_data(appctx);
 	}
 	HA_SPIN_UNLOCK(SFT_LOCK, &sft->lock);