[server][dvc] Invalidate inherited fast-RTS inputs after blob transfer and persist before logging

[sushantmane]

Note on enforce-lines-added: PartitionState schema bump (v22 → v23) is tightly coupled to the consumer code (OffsetRecord.clearInheritedDonorState capture + BlobTransferIngestionHelper.isReplicaLaggedAndNeedBlobTransfer fallback). Splitting into two PRs would land a schema field with no reader, then a reader with no field — neither half is independently verifiable. Adding VALIDATION_OVERRIDE keyword so this can be reviewed end-to-end. The Avro change itself is additive (new field with default: -1), backward-compatible.

VALIDATION_OVERRIDE: schema-and-java-changes-must-land-together-for-this-feature.

Summary

Relationship to the companion PR #2779

#2779 fixes what actually fired in production. The captured incident's RTS promotion went through LeaderFollowerStoreIngestionTask.reportIfCatchUpVersionTopicOffset, observable as the Reported CATCH_UP_BASE_TOPIC_OFFSET_LAG log entry. That path needs cache eviction + an optional leader-complete gate; #2779 adds both.

This PR (#2773) hardens an adjacent path — the fast-RTS code (checkFastReadyToServeForReplica → checkFastReadyToServeWithPreviousTimeLag). That path was not the trigger in the captured incident: PR #2715 already clears the previouslyReadyToServe gate flag in-memory after blob transfer, so today the fast-RTS path correctly bails out at the heartbeat-INVALID guard for a fresh follower (measureHybridHeartbeatTimestamp returns INVALID_MESSAGE_TIMESTAMP while consumedFromUpstream is false).

The risk this PR closes is the crash-restart window: between the moment blob-transfer persists the donor's OffsetRecord to disk and the moment the SIT-side post-blob clear runs, the donor's heartbeatTimestamp and lastCheckpointTimestamp are sitting on disk verbatim. If the JVM crashes in that window, restart re-reads the donor record and the (currentLag - previousLag) ≈ 0 math against the donor's clock would wrongly trigger fast-RTS. Today only the heartbeat-INVALID guard saves us; if that guard ever weakens (refactor, logic bug), the second guard (lastCheckpointTimestamp) is overwritten by now() during normal syncOffset and offers no protection. This PR arms both guards at every observable persistence point.

Both PRs are independent — they touch different methods, fix different classes of artifact, and can land in either order. #2779 is the load-bearing production fix; this PR is belt-and-suspenders for the persistence-window variant.

Builds on top of #2715. After that PR cleared the inherited previouslyReadyToServe gate flag on the post-blob-transfer reinitialized PCS, the OffsetRecord still carried the donor server's heartbeatTimestamp and lastCheckpointTimestamp. checkFastReadyToServeWithPreviousTimeLag would correctly bail out at the heartbeat-INVALID guard for a fresh follower today — measureHybridHeartbeatTimestamp returns INVALID_MESSAGE_TIMESTAMP while consumedFromUpstream is false — but the entire defense relies on that single guard.

If the heartbeat-INVALID early-return is ever bypassed (future refactor of the heartbeat measurement, logic bug, post-restart persistence of mismatched fields), the fast path computes (currentLag - previousLag) ≈ 0 against the donor server's clock and wrongly marks the replica RTS without any real catch-up. The replica is then served reads while reporting INVALID_HEARTBEAT_LAG (Long.MAX_VALUE) for the heartbeat-monitoring metric, which clamps into the metric's top histogram bucket and produces a sentinel artifact in any record-delay / heartbeat-delay SLI dashboard.

Bug call-chain

The fast-RTS path triggers via three different entry points into validateAndSubscribePartition. The post-blob path is the one that wrongly inherits donor state:

StoreIngestionTask.run()                                                 [SIT main loop]
  └─ LFStoreIngestionTask.checkLongRunningTaskState()
       │  // detects partitionConsumptionState.isBlobTransferInProgress() && blobFuture.isDone()
       └─ completeBlobTransferAndSubscribe(pcs)
            │  // adjustStoragePartitionWhenBlobTransferComplete commits SST files
            └─ validateAndSubscribePartition(null, freshPcs)
                 └─ checkConsumptionStateWhenStart(offsetRecord, newPcs)
                      └─ checkFastReadyToServeForReplica(newPcs)
                           │  // gate: hybrid && pcs.getReadyToServeInOffsetRecord()  ← inherited TRUE
                           └─ checkFastReadyToServeWithPreviousTimeLag(pcs)
                                │  previousMessageTimestamp    ← from donor's clock
                                │  previousCheckpointTimestamp ← from donor's clock
                                │  delta = (currentLag - previousLag) ≈ 0
                                └─ pcs.lagHasCaughtUp(); reportCompleted(pcs, true)
                                                                ↑
                                                replica marked ready_to_serve
                                                without any real catch-up

The donor's OffsetRecord lands on disk earlier than this code path ever runs, in P2PMetadataTransferHandler.updateStorePartitionMetadata:

P2PMetadataTransferHandler.handleResponse()                              [blob-transfer thread]
  └─ updateStorePartitionMetadata(storageMetadataService, transferredPartitionMetadata)
       │  OffsetRecord transferredOffsetRecord = new OffsetRecord(transferred bytes)
       └─ storageMetadataService.put(topic, partition, transferredOffsetRecord)  ← donor record persisted verbatim
            └─ StorageEngineMetadataService.put → storageEngine.putPartitionOffset

A JVM crash anywhere between the put above and the SIT-side post-blob clears = restart reads the donor's record and re-triggers fast-RTS via the same gate.

Fix

Close every observable persistence point of the donor's OffsetRecord:

1. At the blob-transfer entry point (the disk window). P2PMetadataTransferHandler.updateStorePartitionMetadata calls OffsetRecord.clearInheritedDonorState(transferredOffsetRecord) immediately after deserializing the wire bytes, so storageMetadataService.put never sees donor-clock values. Crash-safe across the entire SIT-side post-blob sequence.

2. In-memory at the SIT post-blob path (the same-cycle window). completePostTransferPSCUpdated calls OffsetRecord.clearInheritedDonorState(newPcs.getOffsetRecord()) on the record reloaded from storage, so the in-memory PCS used by the immediately-following checkFastReadyToServeForReplica sees cleared values regardless of future changes to the disk-side path.

3. syncOffset(pcs, updateMetadataLag) overload so the post-blob persist path can preserve our cleared timestamps verbatim. The default syncOffset(pcs) keeps current behavior (updateMetadataLag=true); the post-blob path passes false to skip the updateOffsetLagInMetadata re-measurement that would otherwise overwrite lastCheckpointTimestamp with now() and weaken the second early-return guard.

4. OffsetRecord.clearInheritedDonorState(OffsetRecord) is the single authoritative place to extend whenever a new field is added that is unsafe to inherit from a donor host. Today it clears previouslyReadyToServe, heartbeatTimestamp, lastCheckpointTimestamp, and offsetLag. OffsetRecord.PREVIOUSLY_READY_TO_SERVE_KEY is now a public constant; PartitionConsumptionState's private alias is kept for brevity and now references it.

Result: both early-return guards in checkFastReadyToServeWithPreviousTimeLag (heartbeat AND checkpoint) are armed for any post-blob replica at every observable persistence point. Even if a future change weakens one of them, the other still bails out.

Log evidence

Observed on a hybrid AA store (<store>) on a follower replica (<host>) that had just been assigned partitions via Helix LOAD_REBALANCE and bootstrapped via blob transfer. Names sanitized; the timing and structure are real.

05:20:51.92  Helix state-transition: <host>, partition <store>_v<N>_<P>
             received state transition from OFFLINE to STANDBY on session ...
05:20:51.92  HYBRID store current version replica <store>_v<N>-<P>
             initiating transition from OFFLINE to STANDBY ... ST_REBALANCE_TYPE=LOAD_REBALANCE
05:20:59.35  Successfully bootstrapped from blob transfer <store>_v<N>-<P>
             with files: [000390.sst, 000388.sst, ..., MANIFEST-..., CURRENT, OPTIONS-..., ...log]
05:20:59.36  Post-blob-transfer PCS reinitialized for replica: <store>_v<N>-<P> at position: ...
             PCS: PCS{replica=<store>_v<N>-<P>, hybrid=true,
                       latestProcessedVtPosition=Xc{...offset=6334237...},
                       offsetRecord=OffsetRecord{...,
                           heartbeatTimestamp=1777862715170,        ← from donor's clock
                           lastCheckpointTimestamp=1777862722903,   ← from donor's clock
                           ...},
                       leaderFollowerState=STANDBY,
                       leaderCompleteState=LEADER_NOT_COMPLETED,
                       lastLeaderCompleteStateUpdateInMs=0,
                       blobTransferPending=false}
05:20:59.37  Reported STARTED for replica: <store>_v<N>-<P>
05:20:59.37  [Heartbeat lag] replica: <store>_v<N>-<P> is lagging.
             Lag: [9223372036854775807] > Threshold [120000].          ← Long.MAX_VALUE
             Leader Complete State: {LEADER_NOT_COMPLETED}, Last update In Ms: {0}.
05:20:59.53  LeaderCompleteState for replica: <store>_v<N>-<P> changed from
             LEADER_NOT_COMPLETED to LEADER_COMPLETED                  ← ~150ms later, but already RTS

The Long.MAX_VALUE sample lands in the OTel histogram's top bucket and shows up as a P99 spike of ~8.8M ms (~2.4h) on Venice.Server.Ingestion.Replication.{Record,Heartbeat}.Delay — even with Venice.Replica.State=ready_to_serve AND Venice.Version.Role=current filters applied. The replica is structurally RTS via the fast path before it has consumed a single HB.

This change ensures the fast path cannot be entered post-blob regardless of which guard the bug hits.

Testing Done

• Existing SITFastReadyToServeTest continues to pass — non-blob callers see no behavior change (default syncOffset(pcs) keeps updateMetadataLag=true).
• Add unit coverage asserting that after P2PMetadataTransferHandler.updateStorePartitionMetadata, the OffsetRecord persisted to storageMetadataService has heartbeatTimestamp == -1, lastCheckpointTimestamp == -1, offsetLag == DEFAULT_OFFSET_LAG, and the previouslyReadyToServe previousStatuses entry is absent — proves the disk-side window is closed.
• Add unit coverage asserting that completePostTransferPSCUpdated on a freshly reloaded PCS leaves the in-memory OffsetRecord with the same cleared fields after syncOffset(pcs, false) returns — proves the SIT-side window is closed.
• Add unit coverage asserting checkFastReadyToServeWithPreviousTimeLag returns false when both heartbeatTimestamp == -1 and lastCheckpointTimestamp == -1, even with previouslyReadyToServe=true artificially set on the OffsetRecord — proves the second guard is armed.

[Copilot]

Pull request overview

This PR hardens the Da Vinci/Server post–blob-transfer reinitialization flow to prevent inherited (donor) lag metadata in OffsetRecord from incorrectly enabling the fast ready-to-serve (RTS) path after blob bootstrap, and ensures the cleared state is persisted before it is logged.

Changes:

• In completePostTransferPSCUpdated, explicitly invalidate inherited heartbeatTimestamp, lastCheckpointTimestamp, and offsetLag, then persist the updated OffsetRecord before logging.
• Add a syncOffset(PartitionConsumptionState pcs, boolean updateMetadataLag) overload to optionally skip updateOffsetLagInMetadata(...) so explicitly-cleared lag fields can be persisted verbatim.
• Gate the lag-metadata refresh inside syncOffset(...) behind the new updateMetadataLag flag.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 4 out of 4 changed files in this pull request and generated 1 comment.

Comments suppressed due to low confidence (1)

clients/da-vinci-client/src/main/java/com/linkedin/davinci/blobtransfer/client/P2PMetadataTransferHandler.java:131

• Clearing inherited donor state before persisting the transferred OffsetRecord is critical to avoid re-triggering fast RTS after a crash, but there’s no unit coverage validating that the stored OffsetRecord actually has the cleared heartbeatTimestamp/lastCheckpointTimestamp/offsetLag and previouslyReadyToServe removed. Please add a focused unit test for updateStorePartitionMetadata that captures the StorageMetadataService.put() argument and asserts the cleared values are persisted.

        Utils.getReplicaId(transferredPartitionMetadata.topicName, transferredPartitionMetadata.partitionId));
    OffsetRecord transferredOffsetRecord =
        new OffsetRecord(transferredPartitionMetadata.offsetRecord.array(), partitionStateSerializer, null);

    /*
     * Clear inherited donor state before persisting. If donor-clock fields reach disk verbatim, a
     * JVM crash between this put() and the post-blob sync on the SIT thread leaves a state where
     * restart reads donor values and re-triggers fast RTS via checkFastReadyToServeWithPreviousTimeLag.
     * Clearing at the entry point closes that crash window regardless of where the SIT thread is.
     */
    OffsetRecord.clearInheritedDonorState(transferredOffsetRecord);

    // 1. update the offset incremental push job information
    updateIncrementalPushInfoToStore(transferredOffsetRecord, transferredPartitionMetadata);
    // 2. update the offset record in storage service
    storageMetadataService
        .put(transferredPartitionMetadata.topicName, transferredPartitionMetadata.partitionId, transferredOffsetRecord);
    // 3. update the metadata SVS

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 7 out of 7 changed files in this pull request and generated 3 comments.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[misyel]

Is it expected to skip blob transfer for a fresh replica where heartbeat timestamp and donor timestamp is -1? For a new replica, offset record will be initialized with default values -1 and it will also not pass the offset lag check for blob transfer

[sushantmane]

Good catch — let me trace the path explicitly:

For a brand-new replica that has never been through ingestion, the empty OffsetRecord (constructed via getEmptyPartitionState) reports:

• heartbeatTimestamp = -1L (INVALID_MESSAGE_TIMESTAMP)
• donorHeartbeatTimestampMs = -1L (now explicitly initialized in 28925d5 — was 0 before, which is the bug you spotted: 0 would slip past my > 0 capture guard. Fixed.)
• offsetLag = 0 (LOWEST_OFFSET_LAG)
• checkpointedLocalVtPosition = EARLIEST

Walking the method:

1. Time-lag check (lines 153–187): heartbeatTimestamp == INVALID → fall back to donorHeartbeatTimestampMs → also -1 → anchorTimestamp > 0 is false → fall through to offset-lag check.

2. Offset-lag check (lines 188–190): offsetLag == 0 && EARLIEST.equals(checkpointedLocalVtPosition) is the explicit fresh-replica detector — and it returns true (yes, blob transfer is needed).

So the fall-through does land on the right answer for a fresh replica. The path you flagged works because the offset-lag layer has the existing fresh-replica branch. My time-lag layer adds a path before offset-lag for replicas that have ingested before but lost their own heartbeats (post-blob); it deliberately delegates to the offset-lag layer for the no-history case via the > 0 guard.

The donor-init-to--1 fix is in 28925d5 — that closes the latent 0-vs--1 ambiguity you implied. Pinned the contract via testFreshOffsetRecordHasDonorHeartbeatTimestampMsAtSentinel.

[sixpluszero]

I still don't get it, looks like for new or lagged replica this can't be the case. Can we get some concrete example?

[misyel]

Can we also add an integration test to verify fix works E2E?

[sushantmane]

Filing as a separate follow-up rather than holding this PR for it. Reasoning:

The crash-restart window this PR closes is hard to construct deterministically inside an integration test — you need a JVM kill at the precise instant between P2PMetadataTransferHandler.updateStorePartitionMetadata's storageMetadataService.put(...) and the SIT-side post-blob clear, then a restart that re-reads the donor record. Even with a coordinated injection point, the wall-clock delta between donor and recipient is ms-scale, so any non-determinism in test scheduling masks the regression.

What we can assert end-to-end without inducing a crash is the "normal" post-blob path:

• After blob-transfer completes, the persisted OffsetRecord on disk has heartbeatTimestamp == -1, lastCheckpointTimestamp == -1, offsetLag == DEFAULT_OFFSET_LAG, and the previouslyReadyToServe map entry is absent.
• The donor's heartbeat ts is preserved as donorHeartbeatTimestampMs.
• The replica is not prematurely marked READY_TO_SERVE before consuming its own heartbeat.

That's what the unit tests in this PR (testClearInheritedDonorState, testClearInheritedDonorStateIsReentrantSafeOnAlreadyClearedRecord, testShouldStartBlobTransferHybridStoreUsesDonorHeartbeatFallbackPostBlob) already pin down.

For a real E2E test, the most valuable shape would be a multi-server cluster where one replica is bootstrapped via blob transfer from another, and we assert the post-blob OffsetRecord on disk has the cleared sentinels — which exercises both the P2P-thread path and the SIT-thread path. I'll open a follow-up jira / PR for that integration test once the schema bumps and the consumer code in this PR have been in production for a release cycle (so we know the schema is stable). Sound good?

[Copilot]

Pull request overview

Copilot reviewed 9 out of 10 changed files in this pull request and generated 3 comments.

[Copilot]

Pull request overview

Copilot reviewed 9 out of 10 changed files in this pull request and generated no new comments.

[Copilot]

Pull request overview

Copilot reviewed 9 out of 10 changed files in this pull request and generated 1 comment.

[sixpluszero]

If we want to reuse the same v23 before anyone use it, let's still separate a PR with [schema] with this schema change only to keep up with existing case to avoid any confusion. Also this PR let's add one [compact] tag for oncall and any AI bot triage awareness, I hope it can detect it and ask for registration of PS

[sixpluszero]

It looks like correct but over-engineered code. Do we really need it?
Also if existing one is persisted in utf8 and new one write string in java, there is no guarantee that which one will get returned.

[sixpluszero]

I still don't get it, looks like for new or lagged replica this can't be the case. Can we get some concrete example?

[sixpluszero]

I have a general question - if the intention is to avoid the crash restart, I think we can just evict the existing thing and force sync metadata(which your PR already do it) without adding a new field. wdyt?