[aggris2]

Builds on #1007. Until that one is merged, this PR also shows its commit.

A few problems with how the sink finishes, found while testing restarts and failovers:

The final flush prunes undo history. When the stream ends (stopped, or at the stop block), the completion flush passes the stream's latest block as the final block. That deletes the history rows of blocks that are still reversible. If one of them is reorged after the sink restarts, the undo has nothing to revert and the orphaned rows stay. With --undo-buffer-size it also saves the stream's cursor, which is ahead of the blocks still in the buffer, so those get skipped on restart.

A handler error wrapping io.EOF ends the run as if it completed. The stream loop checks errors.Is(err, io.EOF) first, so a lost database connection (failed to begin db transaction: EOF) looks like the stop block was reached. A live sink then exits with status 0 and isn't restarted on failure.

Handler errors are swallowed while stopping. Once the context is cancelled, any error is ignored and the run ends as completed, so the completion handler can flush a block the handler didn't finish.

Also, Revert's deferred rollback never ran: it checked a different err.

Fix:

• The SQL sinker keeps the cursor of the last block it handled and the stream's final block height, and the completion flush uses those. Nothing is flushed if a block failed to be handled.
• The sink library marks handler errors:
  • they never count as end of stream;
  • while stopping, they skip the completion handler, and are reported unless the stop caused them.
• The completion handler gets the cursor of the last block the handler accepted; Sinker.LastReceivedCursor gives the stream's.
• The CLI waits up to 25s for the final flush after a stop, and returns a failure that happens while finishing.

Added TestSinker_CompletionFlush, TestUndoAfterRestart_Postgres (needs PG_DSN), and sink/sinker_stream_test.go, which runs the sinker against a fake stream: completion with an undo buffer, a handler failing while stopping, a handler returning io.EOF. All fail on the base and pass with this change. The existing Postgres integration suites pass.