StreamWAL

[jtanza]

Overall goal

This PR adds StreamWAL, a new per-tablet, leader-only tserver RPC that delivers fully-decoded, committed change events as a stream of CDCSDKProtoRecordPB — the same wire format the existing CDCSDK gRPC connector already consumes.

The client owns all stream state (a per-tablet (term, index) cursor); the server registers nothing. There is no cdc_state table, no stream IDs, no master-driven control plane, and no per-stream aggregation.

The data plane reuses the existing CDCSDK decoder family unchanged: transactional WRITE_OPs are skipped on the wire and the corresponding rows are emitted at APPLYING time by reading intents from IntentsDB, sandwiched between BEGIN/COMMIT envelopes stamped with commit_hybrid_time.

By default StreamWAL delivers records in WAL (apply) order. It also adds an optional, per-request consistent-commit-order mode that instead delivers committed records in commit-time order, watermark-gated, via a composite (term, index, commit_ht) cursor.

StreamWAL lives alongside GetChanges; no existing CDC code path is modified. The one cross-cutting server change is a new wall-clock intent-retention mechanism (gated behind a flag, default-off) that lets a checkpoint-less consumer keep just-applied intents readable.

These changes are broken into a number of sections:

Proto changes

• Adds the StreamWAL RPC to the existing CDCService, three new top-level messages (StreamWalRequestPB, StreamWalResponsePB, StreamWalCursorPB
• Two new optional fields on the existing CDCSDKProtoRecordPB (aborted_subtxn_set, split_tablet_request)
• One new CDCErrorPB::Code value (INTENTS_GC_ERROR = 15).
• Every change is a strictly-additive, backward-compatible proto edit; no existing field numbers or messages change.
• Consistent-commit-order mode adds three optional, additive fields: StreamWalRequestPB.consistent_commit_order (bool, default false), StreamWalCursorPB.commit_ht (the commit-time frontier), and StreamWalResponsePB.resolution_safe_time (the per-tablet resolution watermark the batch gated on).

Files changed:

src/yb/cdc/cdc_service.proto

StreamWAL handler and control flow

• Adds the CDCServiceImpl::StreamWAL handler.
• The driver loop that reads ReplicateMsgs and dispatches each through the decoder, leader/safe-time resolution, INTENTS_GC_ERROR detection, and the partial-APPLYING batch/spill logic.
• The supporting structs (StreamWalDecodeContext, StreamWalIntentResumeState, StreamWalDispatchResult) live in cdc_producer.h.
• All net-new symbols; no existing handler is touched.
• When consistent_commit_order=true, the handler forks early into CDCServiceImpl::HandleStreamWALConsistentCommitOrder and returns; the default (WAL-order) path is left entirely untouched.

Files changed:

src/yb/cdc/cdc_service.cc, src/yb/cdc/cdc_service.h, src/yb/cdc/cdc_producer.h

Decoder helpers reusing the CDCSDK pipeline

• Adds the per-op StreamWAL dispatchers (DispatchWalOpForStreamWAL, DispatchApplyingForStreamWALImpl) and envelope builders (PopulateSyntheticBootstrapDDLs, PopulateStreamWalApplyingRecord, PopulateStreamWalSplitRecord, StampOpIdOnLeadingBeginForStreamWAL).
• These are new callsites layered on top of the existing Populate* decoder family (PopulateCDCSDKWriteRecord, PopulateCDCSDKIntentRecord, DDL/truncate fillers), which are reused verbatim.
• Purely additive the underlying decoders are unchanged, so subtxn-rollback filtering and schema resolution come for free.

Files changed:

src/yb/cdc/cdcsdk_producer.cc

Wall-clock intent retention

• Introduces the runtime flag --intents_min_seconds_to_retain (default 0), a time-based parallel to --log_min_seconds_to_retain for IntentsDB, so a checkpoint-less consumer can read committed-but-just-applied intents without a per-stream lease barrier.
• This is the one area that modifies existing logic rather than only adding: TransactionParticipant::Cleanup and the IntentsDB compaction filter now consult the commit hybrid time when deciding whether to GC a transaction's intents, and TransactionIdApplyOpIdMap in transaction.h changes from mapping to a bare OpId to a TransactionApplyOpIdInfo struct that also carries commit_ht.
• All new retention behavior is gated behind the flag — with the default 0, every path falls through to existing behavior, so this is a no-op for non-CDC and lease-based CDC clusters.
• The remaining files are additive helpers: RunningTransaction::SetApplyHybridTimes (stamps commit/log HT for single-batch applies), Tablet/docdb::CountTxnReverseIndexEntriesForCDC (distinguishes a real intent-GC from a zero-intent txn to avoid spurious INTENTS_GC_ERROR).

Files changed:

src/yb/tablet/transaction_participant.cc, src/yb/docdb/docdb_compaction_filter_intents.cc, src/yb/common/transaction.h, src/yb/tablet/running_transaction.cc, src/yb/tablet/running_transaction.h, src/yb/tablet/tablet.cc, src/yb/tablet/tablet.h, src/yb/docdb/docdb.cc, src/yb/docdb/docdb.h

Stream-less metadata bootstrap

• Adds an idempotent initializer so StreamMetadata can be constructed for stream-id-less callers, letting StreamWAL reuse the existing decoder context without registering a stream.
• Additive; existing stream-backed construction paths are unchanged.

Files changed:

src/yb/cdc/xrepl_stream_metadata.cc, src/yb/cdc/xrepl_stream_metadata.h

Consistent commit-order mode (optional, per-request)

• Default StreamWAL ships records in WAL (apply) order. A multi-shard transaction's rows are emitted at its APPLYING WAL entry, which can materialize arbitrarily later than the commit, so commit_time can move backwards across consecutive records on a tablet.
• With consistent_commit_order=true, the server delivers committed records in non-decreasing commit_time
  order, gated behind the per-tablet resolution watermark, using a composite cursor: (term, index) is the WAL
  re-read floor (and retention point) and commit_ht is the commit-time frontier (the server skips records with
  commit_time <= commit_ht on resume). Floor + frontier together discharge ordering and dedup with zero client
  state.
• Adds GetConsistentChangesForStreamWAL (in cdcsdk_producer.cc) plus the StreamWalConsistentInput /
  StreamWalConsistentOutput structs (in cdc_producer.h). It mirrors the consistent branch of GetChangesForCDCSDK but emits via the existing StreamWAL decoder dispatch and packages the composite cursor.
• Reuse without coupling. The mode is genuinely per-request and independent of the global
  FLAGS_cdc_enable_consistent_records gflag that governs the legacy GetChanges path.

Files changed:

src/yb/cdc/cdcsdk_producer.cc, src/yb/cdc/cdc_producer.h, src/yb/cdc/cdc_service.cc,

Java client bindings & leader-hint failover

• Adds the streamWAL(...) client methods and the request/response wrappers.
• TabletClient.decode gains a CDC-scoped branch that, on LEADER_NOT_READY, applies the tablet_consensus_info hint from the response to refresh the meta-cache leader pointer before retrying (RemoteTablet.applyLeaderHint), avoiding a master round-trip on leader failover mid-stream.
• The hint path only fires for StreamWalRequest/StreamWalResponse pairs, so existing RPC dispatch is unaffected.

Files changed:

java/yb-client/src/main/java/org/yb/client/AsyncYBClient.java, YBClient.java, StreamWalRequest.java (new), StreamWalResponse.java (new), TabletClient.java

Server-side observability metrics

• Adds six per-tablet StreamWAL metrics via a new StreamWALTabletMetrics container. Consistent with the
  stream-id-less design, these attach directly to the tablet's existing metric entity (not a separate
  per-stream entity like CDCSDKTabletMetrics/XClusterTabletMetrics), so they aggregate at the table level
• Lazily instantiated on the first StreamWAL call for a tablet, so tablets that are never streamed cost nothing.
• RPC-level latency / throughput / call-count for StreamWAL come for free from the auto-generated
  handler_latency_yb_cdc_CDCService_StreamWAL family (plus service_request_bytes /
  service_response_bytes), so no custom RPC metrics are added.
• Additive: the only cdc_service.cc touchpoints are net-new — a lazy GetStreamWALTabletMetrics helper,
  streamwal_intents_gc_errors increments on the two INTENTS_GC_ERROR return paths, and the success-path
  gauge/counter updates before RespondSuccess(). No existing metric or handler is modified.

Metrics added:

• streamwal_records_sent — counter (units), aggregated sum. Total decoded change records sent over
  StreamWAL for this tablet.
• streamwal_traffic_sent — counter (bytes), aggregated sum. Total decoded record payload bytes sent over
  StreamWAL.
• streamwal_sent_lag_micros — gauge (µs), aggregated max. Lag between the leader's safe time and the
  commit time of the last record sent.
• streamwal_wal_lag_index — gauge (ops), aggregated max. WAL ops between the leader tip and the read
  cursor (leader_tip.index − next_op_id.index).
• streamwal_intent_retention_window_secs — gauge (s), aggregated max. Current
  --intents_min_seconds_to_retain; lets dashboards derive intent-retention headroom as window − sent_lag.
  A true kMin "headroom" gauge is not expressible (only kSum/kMax aggregation functions exist), so the
  window is surfaced and headroom is computed against the kMax lag downstream.
• streamwal_intents_gc_errors — counter (units), aggregated sum. Count of INTENTS_GC_ERROR responses
  (intents GC'd before StreamWAL could read them). Must always be zero; non-zero indicates data loss.

Files changed:

src/yb/cdc/xrepl_metrics.h, src/yb/cdc/xrepl_metrics.cc, src/yb/cdc/cdc_service.cc,
src/yb/cdc/stream_wal-test.cc

[binks-code-reviewer]

Preserve CDC checkpoint barriers when wall-clock retention is enabled

When --intents_min_seconds_to_retain is nonzero, this branch takes precedence even if GetLatestCheckPointUnlocked() returns a real CDCSDK checkpoint. In a cluster with an existing checkpoint-based CDC stream that enables the new global StreamWAL retention flag, transactions older than the wall-clock window can be inserted for cleanup without checking whether info.apply_op_id is still greater than the stream checkpoint. A lagging lease/checkpoint-based consumer can therefore lose committed intents that were previously protected by the checkpoint barrier.

_{React with 👍/👎 — all feedback helps improve the agent.}

[binks-code-reviewer]

StreamWAL exposes decoded tablet changes without stream authorization

The new StreamWAL handler accepts only a caller-supplied tablet_id and from_op_id before proceeding. In this validation block it checks that those two fields are present, but it does not require a stream_id, does not look up cdc_state, and does not perform an equivalent stream/table authorization check. After this point the handler looks up the tablet, reads WAL entries via ReadReplicatedMessagesInSegmentForCDC, synthesizes streamless metadata over tablet_ptr->metadata()->GetAllColocatedTables(), decodes WAL/intents through DispatchWalOpForStreamWAL, and appends decoded CDCSDKProtoRecordPB records to the response.

The existing GetChanges path is scoped by CDC stream authorization: it requires stream_id/db_stream_id, validates that the requested tablet belongs to that stream, fetches stream metadata from master, and checks stream activity before returning data. This new RPC bypasses that control model entirely. Any client/principal that can reach the tserver CDC service only needs a tablet id to request decoded changes for that tablet. As a result, a client with tserver RPC access could call StreamWAL for a victim tablet and receive decoded row-change data for tables it has not been granted a CDC stream over. This is a broken access-control issue, CWE-862 / OWASP A01.

_{React with 👍/👎 — all feedback helps improve the agent.}

[binks-code-reviewer]

Treat missing post-apply metadata as unsafe under retention

When the compaction filter adds a transaction but does not also see a post-apply marker, info.commit_ht remains invalid and this branch falls through to set.insert(transaction_id). That regresses the existing CDC behavior for unknown apply metadata: committed transactions whose post-apply metadata is absent due to upgrade, crash before asynchronous metadata write, or compaction not including the marker can be routed to abort cleanup even though the retention window cannot be evaluated. Once the participant has evicted the transaction and the coordinator no longer reports it, this can remove committed intents that should have been retained.

_{React with 👍/👎 — all feedback helps improve the agent.}

[binks-code-reviewer]

Do not delete whole intent SSTs based on intent write time

The whole-file retention gate compares the current time to best_file_max_ht, which is the file frontier/newest intent write time rather than the transaction apply time. A long-running transaction can write intents into an old SST, remain open longer than --intents_min_seconds_to_retain, and then commit. Once it is no longer a running transaction, this check can see the old file frontier as outside the window and delete the entire SST immediately, bypassing per-transaction retention logic. StreamWAL can then read the recent APPLYING WAL record but fail to read the committed intents.

_{React with 👍/👎 — all feedback helps improve the agent.}