[SahinurDEV]

Which issue(s) this PR fixes:
Fixes #5619

What this PR does / why we need it:
With require_ack_response and heartbeat_type none, a chunk that the destination never acknowledges is rolled back with update_retry: false and then sent again immediately. retry_timeout, retry_max_times and <secondary> never apply.

For the default heartbeat_type transport, that update_retry: false is intentional: ack_check calls node.disable!, the next write raises NoNodesAvailable, and that starts the retry state. NoneHeartbeatNode#available? always returns true, so disable! does not take that path.

This keeps update_retry: false when disable! actually makes the node unavailable, so a later NoNodesAvailable is not counted twice. When the node is still available after disable! (or there is no node), the rollback now starts the retry state so the retry settings are honored.

Docs Changes:
fluent/fluentd-docs-gitbook#722 (require_ack_response, recover_wait, heartbeat_type none)

Release Note:
out_forward: honor retry and secondary after a failed ACK when heartbeat_type is none

[Watson1978]

Thanks for picking up #5619. Before going into the code, I need to correct the issue I wrote, because this PR builds on it. The issue says that with the default heartbeat_type transport, the next write raises NoNodesAvailable and the retry starts. That depends on timing. An ACK failure disables the node without clearing its failure-detector samples. When the sample window is already large enough, a heartbeat can bring the node back before the next flush. The chunk is then resent without a retry. I added the details to the issue. Sorry about that.

I compared this PR with master (e7d3c3e). For the timeout tests, the destination accepts chunks over TCP but never sends an ACK. Unless noted otherwise, the sender uses require_ack_response true, ack_response_timeout 1, flush_interval 0.5, retry_wait 1 and the default memory buffer. The counts below are from single runs and vary a little between runs.

1. Queued chunks can be discarded even though another node can accept them. heartbeat_type none, two <server>s (one never ACKs, one ACKs normally), retry_max_times 0, 8 chunks with different tags. In 12 seconds, master delivered 6 to the healthy node and kept the other 2 for resending. This PR delivered 3 and discarded 5. With the healthy node as standby, master kept all 8, and this PR discarded all 8. One failed ACK counts as a failure of the whole output, and clear_queue! discards every queued chunk. When a send itself fails, select_healthy_node tries the other available nodes in the current node list first, and raises only when none of them succeeds. (With a file buffer, the chunks are evacuated with a warning instead.)
2. The discard is not logged. With heartbeat_type none, one <server> and retry_max_times 1, the chunk is gone after two failed ACKs. rollback_write(update_retry: true) passes no error to update_retry_state, so neither failed to flush the buffer nor Hit limit for retries is logged. The only trace is buffer queue cleared at debug level.
3. Transport can start a retry depending on timing. If a heartbeat brings the node back between node&.disable! and node.available?, update_retry becomes true for transport too. In a separate test with a closed connection, I forced this interleaving (recover_wait 0, immediate flushing with both flush thread intervals at 0.05, retry_wait 30s). Master resent the chunk after 0.9 seconds. This PR was still waiting for the retry 5 seconds later.
4. Transport can still resend without a retry. This is the same on master. With heartbeat_type transport, recover_wait 10 and retry_max_times 3, sending the first chunk 12 seconds after startup gave 6 failed ACKs and 6 recovered forwarding server lines in 20 seconds, and no retry. With flush_thread_interval 0.05 and flush_thread_burst_interval 0.05, the next flush ran before the heartbeat, and the retry started.

All of this is about failed ACKs, that is, an ACK timeout or a closed connection. The wrong-ACK path (CHUNKID_UNMATCHED) has a similar gap, but I would keep it out of this PR.

I see three directions, and I would like to hear what you think.

• A. Keep counting a retry for heartbeat_type none, and fix problems 1 to 3. The hard part is problem 1. A failed ACK arrives later and for one node, while the retry state belongs to the whole output. Tracking which nodes a chunk has been tried on is one way, but it adds new state. Problem 2 needs the error passed to update_retry_state, or a log of its own. Problem 4 stays.
• B. Keep the node out after a failed ACK, and keep update_retry: false. Writes then go to other nodes, or raise NoNodesAvailable when no node is left, and the usual retry, <secondary> and logging apply. This is the direction I lean toward. I tried an exploratory prototype (diff below). With regular nodes in the cases above, all 8 chunks reached the healthy node, the limit was logged as Hit limit for retries, and transport started the retry. But it is not ready:
  • In the static two-server setup with a standby, the prototype does not promote the standby with heartbeat_type none, because no heartbeat timer rebuilds the node list. All 8 chunks were discarded (with Hit limit logged).
  • A heartbeat that runs between disable! and @failure.clear can bring the node back at once. Also, clearing the samples does not keep the node out for a fixed time, because successful writes in flight add samples too.
  • With recover_wait 0, nothing changes from master.
  • A single transient failure keeps the only node out. When the destination closed the connection once and then sent ACKs normally, delivery took about 17 to 19 seconds with both transport and none, instead of about 2 to 3 seconds on master. There were 4 no nodes are available warnings. With retry_max_times 1, the chunk was discarded before it could be resent.
  • It changes what heartbeat_type none and recover_wait mean, so the docs would need an update.
• C. Always count a failed ACK as a retry, for every heartbeat type. This is the smallest change, but it reverses 2f40abb (#1665). Problems 1 and 2 would then need the same fixes for every heartbeat type, including transport setups that fail over to another node today.

Exploratory prototype for B (on e7d3c3e, with the gaps listed above)

--- a/lib/fluent/plugin/out_forward.rb
+++ b/lib/fluent/plugin/out_forward.rb
@@ -545,7 +545,7 @@ def ack_check(select_interval)
           # instead of returning it to the reuse pool; otherwise a late ack
           # could later be read against a chunk sent on the reused socket (#4057).
           @connection_manager.discard(sock)
-          node&.disable!
+          node&.detach!
           rollback_write(chunk_id, update_retry: false) if chunk_id
         when AckHandler::Result::CHUNKID_UNMATCHED
           # The ack read from this socket belongs to a different chunk than the
@@ -630,6 +630,12 @@ def disable!
         @available = false
       end
 
+      # Clear the failure-detector samples so that recovery needs fresh ones.
+      def detach!
+        disable!
+        @failure.clear
+      end
+
       def standby?
         @standby
       end
@@ -852,7 +858,11 @@ def connect(host = nil, ack: false, &block)
     # Override Node to disable heartbeat
     class NoneHeartbeatNode < Node
       def available?
-        true
+        @detached_until.nil? || Fluent::Clock.now >= @detached_until
+      end
+
+      def detach!
+        @detached_until = Fluent::Clock.now + @sender.recover_wait
       end
 
       def tick

[SahinurDEV]

@Watson1978 Thank you for the detailed review and for testing all of these cases, and for the correction to the issue. I agree with B. I pushed 9bb8423, which builds on your prototype and replaces the earlier update_retry change.

What happens after a failed ACK (timeout or closed connection)

• Another node can take the chunk (another available node with weight > 0, including a standby): the node is detached for recover_wait, the weight array is rebuilt right away, and the chunk is rolled back with update_retry: false. A warning with the chunk id is logged (failed to receive the ack response, detached forwarding server '...' for 10s. the chunk will be sent to another server). No retry is counted and nothing is cleared, so the queued chunks go to the other node.
• It is the last available node, or recover_wait is 0: the node is not detached. The chunk is rolled back, and the retry is counted with an AckResponseError through update_retry_state(chunk_id, using_secondary, error). Then retry_wait, retry_timeout, retry_max_times and <secondary> apply. failed to flush the buffer or Hit limit for retries is logged with that error.

The gaps you listed

1. Discarded chunks while another node is healthy (problem 1): a failed ACK on one node no longer starts the output's retry state while another node can take the chunk, so clear_queue! is not reached in that case.
2. Standby with heartbeat_type none: NoneHeartbeatNode#available? is false while it is detached. The weight array is rebuilt right after the detach, and also before a node is selected in write/try_write when the availability changed (only for none, since it has no heartbeat timer). So the standby is used while the active node is out and is switched back after recover_wait.
3. A heartbeat between disable! and @failure.clear, and samples from in-flight writes: Node#detach! sets a hold time before disable! and clears the samples. heartbeat does not bring the node back until the hold is over, and after that it still needs fresh samples, as before.
4. recover_wait 0: there is no point in detaching a node that would come back at once, so this follows the last-node path and counts a retry.
5. A single transient failure on the only node (the 17-19 s case): the last node is not detached, so the chunk is resent after retry_wait, the same timing as master, and no NoNodesAvailable warnings appear. With retry_max_times 1, the chunk gets its resend.
6. Silent discard (problem 2): the retry now carries an error, so the retry and the discard are logged at warn/error. Detaches are logged as warnings.
7. Transport timing (problems 3 and 4): the choice is made before anything is disabled and no longer depends on node.available? afterwards, so the heartbeat race does not change the result. With a single transport node, a failed ACK now always counts a retry, so it can no longer be resent without one.
8. Docs: I updated the recover_wait description in the plugin. If this direction looks right, I can open a PR to fluentd-docs-gitbook to describe the new behaviour of recover_wait and heartbeat_type none.

The wrong-ACK path (CHUNKID_UNMATCHED) is unchanged, as you suggested.

Tests (test/plugin/test_out_forward.rb)

• a server that never sends the ack (#5619): end-to-end tests of your scenario. There are two in_forward servers, one that accepts chunks but never ACKs and one healthy, with heartbeat_type none, ack_response_timeout 1, retry_max_times 0 and 8 chunks with different tags. One test uses a regular healthy node and the other a standby one. Both check that all 8 tags arrive and that Hit limit for retries is not logged. Both fail on the previous commit of this PR (295f6c4) because chunks are lost.
• a failed ack (#5619): unit tests for both none and transport. They cover the detach plus failover without a retry, the standby being used and then switched back after recover_wait (none), a transport heartbeat not ending the hold early, the last node counting a retry with the error, both nodes failing, and recover_wait 0.

test_out_forward.rb and test/plugin/out_forward/* pass locally (138 tests), except for two DNS tests (configure with ignore_network_errors_at_startup, ... for a primary disabled at startup). Those also fail on master in my environment, because the resolver there answers for any hostname.

I'm happy to change any of this. For example, if you would rather detach the last node with a short hold than count a retry for it.

[Watson1978]

The new tests in a server that never sends the ack (#5619) fail on macOS most of the time. In the Test workflow for 9bb8423, 8 of the 9 macOS jobs across three attempts failed, while all Ubuntu and Windows jobs passed. The failures are in delivers every chunk to the healthy server and delivers every chunk to the healthy standby server. Each one runs for about 31 seconds, and then some chunks are missing from the healthy server. The missing tags differ between runs, from one chunk to six. For example:

<["test.1", "test.2", "test.3", "test.4", "test.5", "test.6", "test.7", "test.8"]> expected but was
<["test.1", "test.2", "test.3", "test.4", "test.5", "test.7", "test.8"]>

These tests check that no chunk is lost. So please find out whether chunks are really lost on macOS, or whether the test does not wait correctly there, and fix the cause. Please don't only make the timeout longer or skip the test on macOS without knowing the cause.

[SahinurDEV]

@Watson1978 Thank you for catching this. The chunks were really dropped on macOS, and the cause is in the ack reader (AckHandler#collect_response), not in how the test waits. I pushed 8a61903.

To find it, I ran the two tests on macOS runners in my fork with the plugin logs dumped on failure. There were two separate causes:

1. An ack that had already arrived was regarded as missing. A socket enqueued while the ack thread is waiting in IO.select is not part of that select call. It is only selected in the next loop, and the timeout is checked before reading. With ack_response_timeout 1 the select interval is also 1s (delayed_commit_timeout / 3), so the healthy server's ack can already be in the socket buffer when it is regarded as timed out. In a failing run, the healthy server responded within 1ms of each send, but no response from node was still logged for it. Because silent had just been detached, healthy was the last node, so a retry was counted and retry_max_times 0 dropped the queue (Hit limit for retries ... from forwarding server 'healthy').
2. A failed ack could be reported too late. Expired acks were only reported after a full select wait, so up to two select intervals after the timeout, which is delayed_commit_timeout (ack_response_timeout + 2). Then Output#try_rollback_write rolled the chunk back first (failed to flush the buffer chunk, timeout to commit) and called update_retry_state without an error. With retry_max_times 0 that cleared the queue without any Hit limit for retries log, which is why some failing runs had no error at all.

Both depend on timing, so they show up more on the slower macOS runners. I couldn't reproduce either one locally on Linux. The first one mainly matters with a short ack_response_timeout, while the second can happen with any value.

The fix: when the timeout is checked, an ack that is already in the socket buffer is read instead (checked with a zero-timeout select), and expired acks are reported without waiting for the select interval. I added two tests to test_ack_handler.rb. I also changed when ack is expired so it no longer writes data to the pipe, because with this change data that is already readable gets read.

Verification: I ran the two tests as they are in the PR 3 times in each of 6 macOS jobs (Ruby 3.3 and 4.0). Before the fix, 13 of 18 runs failed (run). After it, 18 of 18 passed (run). I didn't change any timeout or skip any test.

[Watson1978]

Thanks for the quick and thorough update, and for tracking down the macOS failures.

I compared 8a61903 with master (e7d3c3e) again. The problems from my earlier review are fixed in the cases I measured:

• With heartbeat_type none, two regular servers (one never ACKs) and retry_max_times 0, all 8 chunks reached the healthy server within 12 seconds. The same when the healthy one is a standby, and with transport and two regular servers. In single runs, master delivered 5, 0 (standby) and 6 of them in that time.
• With one server and retry_max_times 1, the chunk is resent once, and then Hit limit for retries is logged with AckResponseError.
• When the only server closes the connection once, the chunk was delivered after about 3 seconds over TCP, close to master (2 to 3 seconds).
• I could reproduce both ack reader causes you described with a forced ordering, and 8a61903 fixes both. The macOS jobs pass now.

Two problems are left, both inline. I reproduced the first one by pausing the ack thread at a fixed point, so I don't know how often it happens in real deployments. The second one happened in every run with a slow destination.

Two more things:

• 8a61903 has no Signed-off-by line, so the DCO check fails. Please add it, for example with git commit --amend --signoff and a force push.
• Yes, please open the docs PR for recover_wait and heartbeat_type none. Please also describe the change for other heartbeat types. When recover_wait is positive and another available server with a positive weight can take the chunks, a server whose ACK fails is now kept out for at least recover_wait (10 seconds by default). On master, a transport heartbeat could bring it back at the next heartbeat.

[Watson1978]

MUST: This rebuilds the weight array from the ack thread. With heartbeat_type none, rebalance_if_availability_changed also rebuilds it from the flush threads. On master, after start, only the event loop rebuilds it (the heartbeat timer and service discovery), so two rebuilds never overlap. Now a rebuild that read the node list earlier can publish its array after a newer one.

I reproduced this with dynamic service discovery and three servers (failed, removed, retained):

1. The ack for failed fails, and this service_discovery_rebalance reads the current services. I paused the ack thread there.
2. Service discovery removes failed and removed, and publishes a weight array with only retained.
3. The ack thread resumes and publishes its array, built from the old list.

After that, master sends only to retained. With this PR, removed is still selected. availability_changed? only checks the current services, so the check before a write does not notice it, and the stale array stays until another rebuild. A lock around the assignment of @weight_array alone would not help, because the node list is read before it.

Please make sure that an older rebuild cannot replace a newer one, and add a test that forces this order. The standby case (delivers every chunk to the healthy standby server) should keep passing.

[SahinurDEV]

Thanks for reproducing this with a forced order. Fixed in 6e6e1eb.

LoadBalancer#rebuild_weight_array now holds a rebuild lock while it reads the node list, builds the array, and publishes it. As you pointed out, locking only the assignment wouldn't help, so out_forward gives the load balancer a nodes_provider (service_discovery_services). Every rebuild reads the list under the lock and ignores the list the caller read earlier. That covers the ack thread, the flush threads, the heartbeat timer, and the service discovery rebuild through Manager#rebalance. A rebuild that read the old list has to publish before the next rebuild can take the lock. Any later rebuild, such as the one service discovery runs after removing failed and removed, then reads the new list and publishes last. The service discovery helper and its rebalance(services) interface are unchanged.

Tests:

• concurrent rebuilds of the weight array (#5627) / a rebuild that read the old node list does not replace a newer one in test_out_forward.rb follows your steps. It uses three servers (failed, removed, retained) and pauses the ack-thread rebuild right after it has read the node list. Meanwhile, another thread sends SERVICE_OUT for failed and removed and calls Manager#run_once. The test then resumes the paused rebuild and checks that the weight array and the selected nodes contain only retained. It fails on 8a61903 (removed is still in the array) and passes now.
• Two unit tests in test_load_balancer.rb check that the provider is used and that the list is read under the lock.
• delivers every chunk to the healthy standby server and the other never sends the ack and a failed ack tests still pass.

[Watson1978]

MUST: A late ack failure can count a second retry for a send attempt that the delayed commit timeout has already handled. When rollback_write(chunk_id) returns true, it does not mean that the failed attempt was rolled back. By then, the same chunk can be in flight again.

I reproduced this without forcing any order. One server, heartbeat_type none, ack_response_timeout 1, retry_wait 1, retry_max_times 1, and one 24 MB chunk in a memory buffer. The destination reads 4 MB per second. It never ACKs the first connection, and ACKs the later ones.

1. The first send of chunk X takes longer than delayed_commit_timeout (ack_response_timeout + 2, 3 seconds here). The ack timeout only starts when the send finishes. So right after the send, Output#try_rollback_write rolls X back and counts one retry (timeout to commit).
2. After retry_wait, X is sent again on a new connection.
3. While X is being sent again, the ack thread reports the ack timeout of the first send. This rollback_write(X) takes back the new attempt and returns true, so a second retry is counted for the same failure.

In three runs each over TCP, master delivered X on the second connection after about 14.5 seconds. With this PR, the second retry reached retry_max_times 1, and X was dropped after about 9 seconds (Hit limit for retries ... AckResponseError). On master, the late failure can also take back the new attempt, but it does not count a retry.

Please don't count a retry here when the delayed commit timeout has already rolled back the failed attempt. A later send of the same chunk can fail on its own. Its ack failure should still count, unless the delayed commit timeout has already rolled back that attempt too. Please add tests for both cases.

[SahinurDEV]

Thanks, you're right: rollback_write returning true doesn't mean the failed attempt was rolled back. Fixed in 0420c88.

Each ack now carries the delayed commit entry (Output::DequeuedChunkInfo) that Output#try_flush added for that send. try_write looks it up and passes it through Node#send_data to the AckHandler, and collect_response yields it back with the result. When the ack fails, on_ack_failed removes that entry while holding @dequeued_chunks_mutex of the instance that flushed the chunk. That is the same lock Output#try_rollback_write holds, so only one of the two can roll the attempt back.

• If the entry is still there, the chunk is rolled back as before and, for the last available node, a retry is counted with AckResponseError.
• If the delayed commit timeout has already removed it, the late failure only logs ... after the delayed commit timeout rolled the chunk back. no retry is counted again. No retry is counted, and the chunk is not taken back either, because the send that is now in flight has its own entry and its own ack. The same check applies when the node is detached.
• A later send of the same chunk has a new entry. So when its ack fails on its own, a retry is counted, unless the delayed commit timeout has already rolled back that attempt too.

Tests are in a late ack failure after the delayed commit timeout (#5627) in test_out_forward.rb. They use the real Output#try_flush, #try_rollback_write, and a memory buffer, with only the network send stubbed:

• does not count a second retry for an attempt that was already rolled back: the first send expires, the chunk is sent again, then the first ack fails. No second retry is counted, and the chunk stays in flight. This fails on 8a61903.
• counts a retry when a later send of the chunk fails on its own
• does not count a retry when the delayed commit timeout rolled back the later send too
• counts a retry for a failed ack before the delayed commit timeout
• passes the delayed commit entry of the send to the ack, plus an AckHandler test that the entry is yielded back.

I also ran your scenario end to end on Linux: one server, heartbeat_type none, ack_response_timeout 1, retry_wait 1, retry_max_times 1, and one 24 MB chunk in a memory buffer. The destination reads about 4 MB/s, never ACKs the first connection, and ACKs the later ones. With 8a61903, the chunk was dropped in both runs (Hit limit for retries ... AckResponseError). With this branch, it was delivered on the second connection after about 12 seconds in both runs, with no Hit limit log.

[SahinurDEV]

@Watson1978 Thanks again for the detailed review. I've replied on both inline threads. The fixes are in 6e6e1eb (weight array rebuilds) and 0420c88 (late ack failure), plus a2a3990 for a RuboCop fix in a test.

About 54cfc28: another attempt at the same two fixes had already been pushed to this branch. It added a generation to Manager#rebalance and called rebalance(services, generation), which changes the service discovery helper's load balancer interface. It also made some private out_forward methods public and kept a list of rolled-back attempts that is never cleaned up. I merged it with the ours strategy (abed3b4), so the history is kept but none of its changes are in the tree. Sorry for the noise.

DCO: I'm sorry, but we can't rewrite the history of this branch, so I won't amend 8a61903 or force-push. 8a61903 ("out_forward: do not regard a received ack as missing in the ack reader") was authored and committed by me, SahinurDEV. For that commit, I certify the Developer Certificate of Origin 1.1 (https://developercertificate.org/) here:

Signed-off-by: SahinurDEV <68515168+SahinurDEV@users.noreply.github.com>

All newer commits on the branch are signed off. Could a maintainer set the DCO check to pass for 8a61903, or squash-merge the PR with this sign-off in the commit message? If you'd prefer another way, please let me know.

Docs: I opened fluent/fluentd-docs-gitbook#722 for require_ack_response, recover_wait and heartbeat_type none. It also covers the change for the other heartbeat types: a server whose ACK fails is now kept out for at least recover_wait while another available server with a positive weight can take its chunks. Before, a transport heartbeat could bring it back at the next heartbeat.

Tests: I ran test_out_forward.rb and test/plugin/out_forward/*.rb locally on Linux three times. The only failures were configure with ignore_network_errors_at_startup and the heartbeat timer switches to the standby node for a primary disabled at startup. Both fail the same way without these changes, because the DNS on my machine resolves any made-up hostname. I also ran the late ack failure, concurrent rebuilds, a failed ack and never sends the ack cases 10 more times, and they passed every time. On macOS (GitHub Actions on my fork, Ruby 3.3 and 4.0, three jobs each), every job ran those cases 5 times and then the whole out_forward suite, and all of them passed: https://github.com/SahinurDEV/fluentd/actions/runs/43208451432 . The PR's CI runs here are waiting for a maintainer to approve them.

[Watson1978]

@SahinurDEV Thanks for the update, and for opening fluent/fluentd-docs-gitbook#722.

About the DCO: Fluentd needs a Signed-off-by line in every commit of a PR (see CONTRIBUTING.md), so we can't merge this PR until the DCO check passes. A sign-off in a comment does not make the check pass. Force-pushing your own PR branch is fine for us.

Please don't use the git rebase HEAD~8 --signoff command from the DCO check page here. Because the ours merge is dropped during the rebase, it applies 54cfc28 again and stops with conflicts. Squashing the branch into one signed-off commit keeps the current tree as it is:

git reset --soft e7d3c3e4
git commit -s
git push --force-with-lease

This also removes the ours merge and the unused 54cfc28 from the history. I'll re-review the changes after the DCO check passes.