Bug: Missed :flushed notification from WriteLoop to Consumer blocks WAL acknowledgment #4058

Summary

Under certain conditions, the {Storage, :flushed, offset} message sent by WriteLoop.flush_buffer/2 at write_loop.ex:329 is lost and never received by the Consumer process. This prevents the FlushTracker from advancing, which blocks WAL acknowledgment to Postgres, causing unbounded WAL retention and replication slot bloat.

Impact

A single stuck shape consumer blocks the global flush offset for the entire stack. All other shapes can be fully caught up, but the FlushTracker's global minimum is pinned to the stuck shape's last acknowledged offset. Postgres WAL accumulates indefinitely until the issue is resolved or the service is restarted.

In the observed incident, 1.72 GiB of WAL backlog accumulated before manual intervention.

Affected Code Path

flush_buffer/2 (write_loop.ex:314-342)
  -> IO.binwrite(json_file, buffer)
  -> :file.datasync(json_file)
  -> send(self(), {Storage, :flushed, last_seen_offset})    <-- message is lost

Consumer.handle_info({Storage, :flushed, offset}, state)    <-- never fires
  -> align_offset_to_txn_boundary/2
  -> ShapeLogCollector.notify_flushed/3

Conditions That Trigger the Bug

The observed Consumer had these characteristics:

• Very small data volume: ~540 bytes written from a single matching operation in a 47,520-op transaction
• Buffer never hit the 64KB inline-flush threshold: the only flush was deferred (scheduled timer or hibernation)
• write_unit: :txn_fragment: fragment-based write mode
• Post-flush state: times_flushed: 1, all WriteLoop offsets aligned, open_files: {nil, nil}, buffer_size: 0

The bug appears to be rare and associated with shapes that have very little streaming data (small enough that the buffer is only flushed during hibernation, not inline during writes).

Hypotheses for Root Cause

Hypothesis 1: Message lost during OTP hibernate transition (most likely)

The Consumer's handle_info(:timeout, ...) calls Storage.hibernate() -> flush_and_close_all -> flush_buffer, which calls send(self(), {Storage, :flushed, offset}). The callback then returns {:noreply, state, :hibernate}.

The BEAM's erlang:hibernate/3 reclaims the process heap. While the message should survive in the mailbox (messages are stored outside the heap), there may be an edge case:

• The send(self(), ...) is executed during the callback. The message is placed in the process's own mailbox.
• The callback returns :hibernate. The OTP gen_server loop calls proc_lib:hibernate/3.
• If there is any window between the message being enqueued and the process entering the hibernate wait state where the message reference could be invalidated, the message could be lost.

This would be an OTP/BEAM-level bug and would be extremely rare, which matches our observation of it happening to only one shape.

Hypothesis 2: Scheduled flush timer cancelled before firing

1. append_fragment_to_log! writes data to the buffer and returns {acc, schedule_flush: N}.
2. A flush timer is scheduled for 1 second.
3. Before the timer fires, another event arrives that triggers a flush with cancel_flush_timer: true (if the buffer was drained during that write).
4. The timer is cancelled at pure_file_storage.ex:1208-1209.
5. The inline flush during the second write sends :flushed with a different offset (from the second write).
6. When the Consumer processes this :flushed, align_offset_to_txn_boundary drops entries with offsets less than the flushed offset, silently consuming the original mapping entry without ever calling notify_flushed for its boundary.

However, this hypothesis is weakened by the Consumer's state showing only one txn_offset_mapping entry and no evidence of subsequent writes.

Hypothesis 3: Race with signal_txn_commit!

In the write_unit: :txn_fragment path, append_fragment_to_log! and signal_txn_commit! are called separately. If the scheduled flush fires between these two calls (while the Consumer is processing the commit fragment in maybe_complete_pending_txn):

1. append_fragment_to_log! writes data, schedules a flush timer.
2. The handle_call for the commit fragment begins. The Consumer is processing synchronously.
3. The flush timer fires, but since the Consumer is in a handle_call, the timer message waits in the mailbox.
4. signal_txn_commit! runs, updates last_seen_txn_offset.
5. maybe_complete_pending_txn adds the txn_offset_mapping entry.
6. handle_call returns.
7. The timer message is processed: perform_scheduled_flush -> flush_buffer -> sends :flushed.
8. :flushed is processed -> align_offset_to_txn_boundary runs.

This sequence should still work correctly because step 7 and 8 happen in order. Unless there's a state inconsistency where flush_buffer sees stale offsets.

Hypothesis 4: flush_buffer no-op on empty buffer after prior drain

If the buffer was somehow drained by a path that does NOT send :flushed (e.g. an ETS-only write, or a code path that clears the buffer without going through flush_buffer), then subsequent calls to flush_buffer would match the buffer_size: 0 clause at write_loop.ex:310-312 and return without sending the notification.

The flush_and_close_all at write_loop.ex:78-79 unconditionally calls flush_buffer. If the buffer was already empty, this is a no-op -- no :flushed message is sent. The only flush that occurred (the one that set times_flushed: 1) would have sent :flushed, but if that message was lost or preempted, no subsequent flush would ever re-send it.

This is the most mechanically plausible explanation: once the :flushed message is lost for any reason, there is no recovery path -- no code ever re-sends it.

Proposed Solutions

Fix 1: Synchronous flush notification during hibernation (targeted fix)

Instead of relying on the async send(self(), ...) during flush_and_close_all, have the Consumer's hibernate path call notify_flushed directly after Storage.hibernate() returns:

# In Consumer.handle_info(:timeout, ...)
def handle_info(:timeout, state) do
  if consumer_suspend_enabled?(state) and consumer_can_suspend?(state) do
    {:stop, ShapeCleaner.consumer_suspend_reason(), state}
  else
    state = %{state | writer: ShapeCache.Storage.hibernate(state.writer)}
    # Drain any pending flush notifications synchronously
    state = flush_pending_notifications(state)
    {:noreply, state, :hibernate}
  end
end

Where flush_pending_notifications checks if the WriteLoop has advanced past the last notified offset and calls notify_flushed directly.

Fix 2: Return flush offset from flush_and_close_all (structural fix)

Change flush_and_close_all to return the flushed offset so the caller can act on it synchronously, rather than relying on a self-sent message:

def flush_and_close_all(acc, state) do
  {acc, flushed_offset} = flush_buffer_returning_offset(acc, state)
  # ... close files ...
  {writer_acc(acc, open_files: open_files()), flushed_offset}
end

The Consumer (or Storage.hibernate) would then handle the notification inline.

Fix 3: Periodic flush consistency check (defense in depth)

Add a periodic check (e.g. on the hibernate timeout or a dedicated timer) that compares the WriteLoop's last_persisted_offset against the last offset reported to the FlushTracker. If they diverge, re-send the :flushed notification. This acts as a self-healing mechanism regardless of the root cause.

Fix 4: Decouple flush_buffer notification from send(self(), ...) (cleanest fix)

Instead of having flush_buffer use send(self(), ...) as a side effect buried inside a pure-looking function, have it return the flushed offset as part of its return value. All callers would then be responsible for propagating the notification, making the data flow explicit and eliminating the reliance on message passing within a single process:

def flush_buffer(writer_acc(buffer_size: 0) = acc, _state) do
  {acc, _flushed_offset = nil}
end

def flush_buffer(acc, state) do
  # ... write + fsync ...
  {updated_acc, _flushed_offset = last_seen_offset}
end

This makes it impossible for the notification to be lost, since it's part of the function's return value, not a side-effecting message.