Make MarkRevealed and RecordWonBid order-independent

pkg/payload_bidder/payment_tracker.go

@@ -38,7 +38,19 @@ type PaymentTracker struct {
 	// Pending payments: unrevealed won bids, pending for 2 epochs.
 	// Only these count as "pending" in the UI and for topup checks.
 	pendingPayments map[phase0.Slot]*PendingPayment
-	pendingMu       sync.Mutex
+
+	// earlyReveals records a slot whose MarkRevealed call arrived before
+	// RecordWonBid created its pending entry: the two are fed by independent
+	// goroutines (RevealService's own gate/timer vs. InclusionTracker's
+	// head-event loop) with no happens-before edge between them, so either
+	// order is possible whenever a head event is delayed past the reveal
+	// gate. Without this, that ordering silently drops the balance deduction
+	// (MarkRevealed finds no pending entry and no-ops) and RecordWonBid then
+	// creates a pending entry that is never marked revealed, orphaned until
+	// PruneExpiredPayments clears it ~2 epochs later. Guarded by pendingMu
+	// (the two are always read/written together).
+	earlyReveals map[phase0.Slot]phase0.Epoch
+	pendingMu    sync.Mutex
 
 	chainSvc chain.Service
 	log      logrus.FieldLogger
@@ -48,18 +60,37 @@ type PaymentTracker struct {
 func NewPaymentTracker(chainSvc chain.Service, log logrus.FieldLogger) *PaymentTracker {
 	return &PaymentTracker{
 		pendingPayments: make(map[phase0.Slot]*PendingPayment, 16),
+		earlyReveals:    make(map[phase0.Slot]phase0.Epoch, 4),
 		chainSvc:        chainSvc,
 		log:             log.WithField("component", "payment-tracker"),
 	}
 }
 
-// RecordWonBid records a won bid as a pending payment (unrevealed).
-// Called when our bid is included in a beacon block.
-// If we later reveal, call MarkRevealed to move it from pending to a balance deduction.
-// If we don't reveal, it stays pending for 2 epochs then expires.
+// RecordWonBid records a won bid as a pending payment (unrevealed), unless
+// MarkRevealed already ran for this slot (see earlyReveals) — in that case
+// the deduction that was deferred for lack of a known value applies now,
+// immediately, and no pending entry is created; a bid that was already
+// revealed is never "pending".
+// Called when our bid is included in a beacon block. If we later reveal
+// (the common order), call MarkRevealed to move it from pending to a
+// balance deduction. If we don't reveal, it stays pending for 2 epochs then
+// expires.
 func (t *PaymentTracker) RecordWonBid(slot phase0.Slot, value uint64) {
 	t.pendingMu.Lock()
-	defer t.pendingMu.Unlock()
+
+	if _, revealedEarly := t.earlyReveals[slot]; revealedEarly {
+		delete(t.earlyReveals, slot)
+		t.pendingMu.Unlock()
+
+		t.deduct(slot, value)
+
+		t.log.WithFields(logrus.Fields{
+			"slot":  slot,
+			"value": value,
+		}).Info("Won bid was already revealed before it was recorded: deducted from live balance")
+
+		return
+	}
 
 	epoch := t.chainSvc.GetEpochOfSlot(slot)
 
@@ -69,40 +100,56 @@ func (t *PaymentTracker) RecordWonBid(slot phase0.Slot, value uint64) {
 		Value: value,
 	}
 
+	t.pendingMu.Unlock()
+
 	t.log.WithFields(logrus.Fields{
 		"slot":  slot,
 		"epoch": epoch,
 		"value": value,
 	}).Info("Recorded won bid as pending payment")
 }
 
-// MarkRevealed moves a won bid from pending to an immediate balance deduction.
-// The payment is removed from pending and subtracted from the balance adjustment.
+// MarkRevealed moves a won bid from pending to an immediate balance
+// deduction. If RecordWonBid hasn't run yet for this slot (no pending entry
+// exists), the value isn't known yet, so the deduction can't be applied
+// here: the slot is recorded in earlyReveals instead, and RecordWonBid
+// applies the deduction immediately once it does run.
 func (t *PaymentTracker) MarkRevealed(slot phase0.Slot) {
 	t.pendingMu.Lock()
+
 	p, ok := t.pendingPayments[slot]
 	if !ok {
+		t.earlyReveals[slot] = t.chainSvc.GetEpochOfSlot(slot)
 		t.pendingMu.Unlock()
+
+		t.log.WithField("slot", slot).Warn(
+			"Reveal completed before the won bid was recorded — deferring the balance deduction")
+
 		return
 	}
 
 	value := p.Value
 	delete(t.pendingPayments, slot)
 	t.pendingMu.Unlock()
 
-	// Deduct from live balance, anchored to this slot's epoch so the
-	// reconciler keeps the delta until the snapshot advances past it.
-	t.adjustmentMu.Lock()
-	t.balanceAdjustment -= int64(value)
-	t.anchorEpochLocked(t.chainSvc.GetEpochOfSlot(slot))
-	t.adjustmentMu.Unlock()
+	t.deduct(slot, value)
 
 	t.log.WithFields(logrus.Fields{
 		"slot":  slot,
 		"value": value,
 	}).Info("Revealed bid: deducted from live balance")
 }
 
+// deduct applies a revealed bid's value to the live balance adjustment,
+// anchored to the slot's epoch so the reconciler keeps the delta until the
+// snapshot advances past it.
+func (t *PaymentTracker) deduct(slot phase0.Slot, value uint64) {
+	t.adjustmentMu.Lock()
+	t.balanceAdjustment -= int64(value)
+	t.anchorEpochLocked(t.chainSvc.GetEpochOfSlot(slot))
+	t.adjustmentMu.Unlock()
+}
+
 // AddDeposit credits a deposit/topup to the live balance adjustment, anchored
 // to the current epoch. The credit is reconciled away by ReconcileToEpoch once
 // the authoritative snapshot advances past that epoch.
@@ -194,7 +241,11 @@ func (t *PaymentTracker) SetPaymentDisputed(slot phase0.Slot, disputed bool) {
 	}
 }
 
-// PruneExpiredPayments removes pending payments older than 2 epochs.
+// PruneExpiredPayments removes pending payments older than 2 epochs, and
+// earlyReveals entries of the same age whose matching RecordWonBid never
+// arrived (the won-bid report was itself lost, or the win didn't pan out) —
+// otherwise a slot revealed early but never recorded would sit in
+// earlyReveals forever.
 func (t *PaymentTracker) PruneExpiredPayments(currentEpoch phase0.Epoch) {
 	t.pendingMu.Lock()
 	defer t.pendingMu.Unlock()
@@ -211,4 +262,16 @@ func (t *PaymentTracker) PruneExpiredPayments(currentEpoch phase0.Epoch) {
 			delete(t.pendingPayments, slot)
 		}
 	}
+
+	for slot, epoch := range t.earlyReveals {
+		if currentEpoch > epoch+1 {
+			t.log.WithFields(logrus.Fields{
+				"slot":          slot,
+				"reveal_epoch":  epoch,
+				"current_epoch": currentEpoch,
+			}).Debug("Pruning stale early-reveal marker (matching won bid never recorded)")
+
+			delete(t.earlyReveals, slot)
+		}
+	}
 }

pkg/payload_bidder/payment_tracker_test.go

@@ -88,6 +88,78 @@ func TestPaymentTracker_ReconcileToEpoch(t *testing.T) {
 	assert.Equal(t, int64(0), tracker.GetBalanceAdjustment())
 }
 
+// TestPaymentTracker_MarkRevealedBeforeRecordWonBid guards NM-07: MarkRevealed
+// and RecordWonBid are fed by independent goroutines with no happens-before
+// edge between them, so a delayed head event can deliver them in either
+// order. Before the fix, a MarkRevealed that arrived first found no pending
+// entry and silently no-op'd, permanently dropping the deduction; the later
+// RecordWonBid then created an orphaned pending entry that was never marked
+// revealed.
+func TestPaymentTracker_MarkRevealedBeforeRecordWonBid(t *testing.T) {
+	tracker := newTestPaymentTracker()
+
+	const slot = phase0.Slot(200)
+	const value = uint64(5000)
+
+	// The reveal completes before the win is even recorded.
+	tracker.MarkRevealed(slot)
+	assert.Equal(t, int64(0), tracker.GetBalanceAdjustment(),
+		"the value isn't known yet, so no deduction can apply until RecordWonBid runs")
+	assert.Equal(t, uint64(0), tracker.GetTotalPendingPayments())
+
+	// The delayed head event now lands and the win is recorded.
+	tracker.RecordWonBid(slot, value)
+
+	// The deduction lands immediately -- the bid is never treated as
+	// "pending" despite arriving after the reveal, since it was already
+	// revealed.
+	assert.Equal(t, -int64(value), tracker.GetBalanceAdjustment(),
+		"NM-07: the deduction must still apply once the won bid is recorded")
+	assert.Equal(t, uint64(0), tracker.GetTotalPendingPayments(),
+		"a bid revealed before it was recorded must never sit in pending")
+}
+
+// TestPaymentTracker_RecordWonBidBeforeMarkRevealed confirms the common
+// ordering (win recorded, then revealed) is unaffected by the NM-07 fix.
+func TestPaymentTracker_RecordWonBidBeforeMarkRevealed(t *testing.T) {
+	tracker := newTestPaymentTracker()
+
+	const slot = phase0.Slot(201)
+	const value = uint64(3000)
+
+	tracker.RecordWonBid(slot, value)
+	assert.Equal(t, uint64(value), tracker.GetTotalPendingPayments())
+	assert.Equal(t, int64(0), tracker.GetBalanceAdjustment())
+
+	tracker.MarkRevealed(slot)
+	assert.Equal(t, uint64(0), tracker.GetTotalPendingPayments())
+	assert.Equal(t, -int64(value), tracker.GetBalanceAdjustment())
+}
+
+// TestPaymentTracker_EarlyRevealNeverRecordedIsPruned confirms a slot whose
+// reveal arrived early but whose matching RecordWonBid never shows up (the
+// won-bid report was lost, or the win didn't pan out) doesn't leak forever:
+// PruneExpiredPayments clears it the same way it clears an expired pending
+// payment.
+func TestPaymentTracker_EarlyRevealNeverRecordedIsPruned(t *testing.T) {
+	tracker := newTestPaymentTracker()
+
+	tracker.MarkRevealed(32) // epoch 1, per stubChainService's slot/32 mapping
+	assert.Len(t, tracker.earlyReveals, 1)
+
+	tracker.PruneExpiredPayments(phase0.Epoch(2))
+	assert.Len(t, tracker.earlyReveals, 1, "must stay through payment epoch + 1")
+
+	tracker.PruneExpiredPayments(phase0.Epoch(3))
+	assert.Empty(t, tracker.earlyReveals)
+
+	// A RecordWonBid arriving after the marker was pruned falls back to the
+	// normal pending path rather than (wrongly) deducting immediately.
+	tracker.RecordWonBid(32, 100)
+	assert.Equal(t, uint64(100), tracker.GetTotalPendingPayments())
+	assert.Equal(t, int64(0), tracker.GetBalanceAdjustment())
+}
+
 func TestPaymentTracker_PruneExpiredPayments(t *testing.T) {
 	tracker := newTestPaymentTracker()