diff --git a/internal/blocklistener/blocklistener.go b/internal/blocklistener/blocklistener.go index e49b32ae..df293af5 100644 --- a/internal/blocklistener/blocklistener.go +++ b/internal/blocklistener/blocklistener.go @@ -42,17 +42,22 @@ func BufferChannel(ctx context.Context, target NewBlockHashConsumer) (buffered c go func() { defer close(done) var blockedUpdate *ffcapi.BlockHashEvent + var blockedDiscardCount int for { if blockedUpdate != nil { select { case blockUpdate := <-buffered: // Have to discard this blockedUpdate.GapPotential = true // there is a gap for sure at this point - log.L(ctx).Debugf("Blocked event stream missed new block event: %v", blockUpdate.BlockHashes) + blockedDiscardCount++ + log.L(ctx).Debugf("Blocked event stream missed new block event headBlock=%d discardCount=%d hashes=%v", + blockUpdate.HeadBlockNumber, blockedDiscardCount, blockUpdate.BlockHashes) case target.GetReceiveChannel() <- blockedUpdate: // We're not blocked any more - log.L(ctx).Infof("Event stream block-listener unblocked") + log.L(ctx).Infof("Event stream block-listener unblocked headBlock=%d discardedWhileBlocked=%d", + blockedUpdate.HeadBlockNumber, blockedDiscardCount) blockedUpdate = nil + blockedDiscardCount = 0 case <-ctx.Done(): log.L(ctx).Debugf("Block listener exiting (previously blocked)") return @@ -68,7 +73,8 @@ func BufferChannel(ctx context.Context, target NewBlockHashConsumer) (buffered c // all good, we passed it on default: // we can't deliver it immediately, we switch to blocked mode - log.L(ctx).Infof("Event stream block-listener became blocked") + log.L(ctx).Infof("Event stream block-listener became blocked headBlock=%d downstreamQueueFull=true", + update.HeadBlockNumber) // Take a copy of the block update, so we can modify (to mark a gap) without affecting other streams var bu = *update blockedUpdate = &bu diff --git a/internal/confirmations/confirmations.go b/internal/confirmations/confirmations.go index b66f2844..46e511b9 100644 --- a/internal/confirmations/confirmations.go +++ b/internal/confirmations/confirmations.go @@ -62,12 +62,13 @@ const ( ) type Notification struct { - NotificationType NotificationType - Event *EventInfo // NewEventLog, RemovedEventLog - Transaction *TransactionInfo // NewTransaction, RemovedTransaction - RemovedListener *RemovedListenerInfo // ListenerRemoved - pending *pendingItem // receiptArrived - receipt *ffcapi.TransactionReceiptResponse // receiptArrived + NotificationType NotificationType + Event *EventInfo // NewEventLog, RemovedEventLog + Transaction *TransactionInfo // NewTransaction, RemovedTransaction + RemovedListener *RemovedListenerInfo // ListenerRemoved + pending *pendingItem // receiptArrived + receipt *ffcapi.TransactionReceiptResponse // receiptArrived + receiptGeneration uint64 // receiptArrived } type EventInfo struct { @@ -156,6 +157,8 @@ type pendingItem struct { previousConfirmationCount *uint64 // headBlockNumber mode: last dispatched CurrentConfirmationCount queuedStale *list.Element // protected by receiptChecker mux lastReceiptCheck time.Time // protected by receiptChecker mux + receiptGeneration uint64 // protected by receiptChecker mux, incremented on each schedule + appliedReceiptGeneration uint64 // last receipt generation applied by confirmationsListener receiptCallback func(ctx context.Context, receipt *ffcapi.TransactionReceiptResponse) confirmationsCallback func(ctx context.Context, notification *apitypes.ConfirmationsNotification) transactionHash string @@ -237,8 +240,12 @@ func (bcm *blockConfirmationManager) Stop() { if bcm.done != nil { bcm.cancelFunc() bcm.receiptChecker.close() - bcm.receiptChecker = nil + // Wait for confirmationsListener to actually observe ctx.Done() and return before + // nil-ing out receiptChecker - its select also has newBlockHashEvents/bcmNotifications + // cases, so a buffered event can still win a race against ctx.Done() and drive one more + // loop iteration that calls into receiptChecker (e.g. via scheduleReceiptChecks). <-bcm.done + bcm.receiptChecker = nil bcm.done = nil // Reset context ready for restart bcm.ctx, bcm.cancelFunc = context.WithCancel(bcm.baseContext) @@ -277,6 +284,13 @@ func (bcm *blockConfirmationManager) Notify(n *Notification) error { return i18n.NewError(bcm.ctx, tmmsgs.MsgInvalidConfirmationRequest, n) } } + queueCap := cap(bcm.bcmNotifications) + queueLen := len(bcm.bcmNotifications) + if queueLen >= queueCap { + log.L(bcm.ctx).Warnf("Confirmation notification queue full (%d/%d), blocking on type=%s", queueLen, queueCap, n.NotificationType) + } else if queueLen >= queueCap-1 { + log.L(bcm.ctx).Warnf("Confirmation notification queue nearly full (%d/%d) type=%s", queueLen, queueCap, n.NotificationType) + } select { case bcm.bcmNotifications <- n: bcm.metricsEmitter.RecordNotificationQueueingMetrics(bcm.ctx, string(n.NotificationType), time.Since(startTime).Seconds()) @@ -374,6 +388,12 @@ func (bcm *blockConfirmationManager) confirmationsListener() { if bhe.GapPotential { bcm.blockListenerStale = true } + blockQueueDepth := len(bcm.newBlockHashEvents) + blockQueueCap := cap(bcm.newBlockHashEvents) + if blockQueueDepth >= blockQueueCap-1 { + log.L(bcm.ctx).Warnf("Confirmation block event queue nearly full (%d/%d) headBlock=%d gapPotential=%t", + blockQueueDepth, blockQueueCap, bhe.HeadBlockNumber, bhe.GapPotential) + } blockHashes = append(blockHashes, bhe.BlockHashes...) bcm.headBlockNumber = bhe.HeadBlockNumber // always update the head block number, NOTE: the number can decrease during a re-org // Need to also pass this event to any confirmed block listeners @@ -403,6 +423,13 @@ func (bcm *blockConfirmationManager) confirmationsListener() { } } startTime := time.Now() + bcm.pendingMux.Lock() + pendingItemCount := len(bcm.pending) + bcm.pendingMux.Unlock() + blockHashCount := len(blockHashes) + notificationCount := len(notifications) + log.L(bcm.ctx).Debugf("Confirmation listener iteration starting trigger=%s blockHashes=%d notifications=%d pendingItems=%d headBlock=%d blockQueueDepth=%d notificationQueueDepth=%d", + triggerType, blockHashCount, notificationCount, pendingItemCount, bcm.headBlockNumber, len(bcm.newBlockHashEvents), len(bcm.bcmNotifications)) // Each time round the loop we need to have a consistent view of the chain. // This view must not add later blocks (by number) in, or change the hash of blocks, @@ -421,14 +448,12 @@ func (bcm *blockConfirmationManager) confirmationsListener() { bcm.blockListenerStale = false } - blockHashCount := len(blockHashes) + newBlockEvent := triggerType == "newBlockHashes" // Process each new block - bcm.processBlockHashes(blockHashes) + bcm.processBlockHashes(blockHashes, newBlockEvent) // Truncate the block hashes now we've processed them blockHashes = blockHashes[:0] - notificationCount := len(notifications) - // Process any new notifications - we do this at the end, so it can benefit // from knowing the latest highestBlockSeen // This returns the notifications that were not processed successfully so we store them @@ -439,24 +464,43 @@ func (bcm *blockConfirmationManager) confirmationsListener() { continue } scheduleAllTxReceipts := !receivedFirstBlock && blockHashCount > 0 + if bcm.chainTrackingMode == ffcapi.ChainTrackingModeLight { + // in light mode, we need to schedule all transactions if we have received any blocks + // this is because in light mode, we do not have block details available to check transaction hashes against + scheduleAllTxReceipts = scheduleAllTxReceipts || newBlockEvent + } // Mark receipts stale after duration bcm.scheduleReceiptChecks(scheduleAllTxReceipts) receivedFirstBlock = receivedFirstBlock || blockHashCount > 0 - log.L(bcm.ctx).Tracef("[TimeTrace] Confirmation listener processed %d block hashes and %d notifications in %s, trigger type: %s", blockHashCount, notificationCount, time.Since(startTime), triggerType) - + iterationDuration := time.Since(startTime) + log.L(bcm.ctx).Debugf("Confirmation listener iteration complete trigger=%s duration=%s blockHashes=%d notifications=%d pendingItems=%d headBlock=%d blockQueueDepth=%d notificationQueueDepth=%d", + triggerType, iterationDuration, blockHashCount, notificationCount, pendingItemCount, bcm.headBlockNumber, len(bcm.newBlockHashEvents), len(bcm.bcmNotifications)) + log.L(bcm.ctx).Tracef("[TimeTrace] Confirmation listener processed %d block hashes and %d notifications in %s, trigger type: %s", blockHashCount, notificationCount, iterationDuration, triggerType) } } -func (bcm *blockConfirmationManager) scheduleReceiptChecks(receivedBlocksFirstTime bool) { +func (bcm *blockConfirmationManager) scheduleReceiptChecks(scheduleUnreceiptedItems bool) { now := time.Now() + // In light mode there is no way to know which block (if any) will contain a given pending + // transaction - unlike full mode, which actively detects a mined transaction by scanning each + // new block's transaction list (processBlock). So in light mode, an item that still has no + // receipt must be retried on every new block, not just scheduled once - otherwise a "not found" + // result on the first check would fall all the way back to the (much slower) stale-timeout for + // every subsequent attempt. This is deliberately restricted to light mode: applying it in full + // mode as well would race against processBlock's own scheduling of the same item (verified by + // reproducing a duplicate in-flight receipt check against TestBlockConfirmationManagerE2ETransactionMovedFork). + retryUnreceiptedLightModeItems := scheduleUnreceiptedItems && bcm.chainTrackingMode == ffcapi.ChainTrackingModeLight for _, pending := range bcm.pending { // For efficiency we do a dirty read on the receipt check time before going into the locking // check within the receipt checker if pending.pType == pendingTypeTransaction { - if receivedBlocksFirstTime && !pending.scheduledAtLeastOnce { + switch { + case scheduleUnreceiptedItems && !pending.scheduledAtLeastOnce: + bcm.receiptChecker.schedule(pending, false) + case retryUnreceiptedLightModeItems && pending.blockHash == "": bcm.receiptChecker.schedule(pending, false) - } else if now.Sub(pending.lastReceiptCheck) > bcm.staleReceiptTimeout { + case now.Sub(pending.lastReceiptCheck) > bcm.staleReceiptTimeout: // schedule stale receipt checks bcm.receiptChecker.schedule(pending, true /* suspected timeout - prompts re-check in the lock */) } @@ -489,7 +533,7 @@ func (bcm *blockConfirmationManager) processNotifications(notifications []*Notif case RemovedTransaction: bcm.removeItem(n.transactionPendingItem(), true) case receiptArrived: - bcm.dispatchReceipt(n.pending, n.receipt, blocks) + bcm.dispatchReceipt(n.pending, n.receipt, n.receiptGeneration, blocks) default: // Note that streamStopped is handled in the polling loop directly log.L(bcm.ctx).Warnf("Unexpected notification type: %s", n.NotificationType) @@ -501,7 +545,17 @@ func (bcm *blockConfirmationManager) processNotifications(notifications []*Notif return notifications[:0], nil } -func (bcm *blockConfirmationManager) dispatchReceipt(pending *pendingItem, receipt *ffcapi.TransactionReceiptResponse, blocks *blockState) { +// NOTE: there is no locking in this function +// relies on the consumer logic to not call this function concurrently +func (bcm *blockConfirmationManager) dispatchReceipt(pending *pendingItem, receipt *ffcapi.TransactionReceiptResponse, receiptGeneration uint64, blocks *blockState) { + if receiptGeneration > 0 && receiptGeneration <= pending.appliedReceiptGeneration { + log.L(bcm.ctx).Debugf("Ignoring stale receipt for transaction %s (actual_generation=%d applied_generation=%d)", + pending.transactionHash, receiptGeneration, pending.appliedReceiptGeneration) + return + } + if receiptGeneration > 0 { + pending.appliedReceiptGeneration = receiptGeneration + } pending.blockNumber = receipt.BlockNumber.Uint64() pending.blockHash = receipt.BlockHash log.L(bcm.ctx).Infof("Receipt for transaction %s downloaded. BlockNumber=%d BlockHash=%s", pending.transactionHash, pending.blockNumber, pending.blockHash) @@ -555,9 +609,13 @@ func (bcm *blockConfirmationManager) removeItem(pending *pendingItem, stale bool bcm.pendingMux.Unlock() } -func (bcm *blockConfirmationManager) processBlockHashes(blockHashes []string) { +func (bcm *blockConfirmationManager) processBlockHashes(blockHashes []string, newBlockEvent bool) { if bcm.chainTrackingMode == ffcapi.ChainTrackingModeLight { // for light chain tracking mode, no block details are available, only need to calculate the number of confirmations using head block number + if !newBlockEvent { + // if this function was not triggered by a new block event, we do not need to process any pending transactions + return + } bcm.checkAndDispatchConfirmationsUsingBlockHeight() return } @@ -841,7 +899,9 @@ func (bcm *blockConfirmationManager) checkAndDispatchConfirmationsUsingBlockHeig for _, p := range bcm.pending { items = append(items, p) } + headBlock := bcm.headBlockNumber bcm.pendingMux.Unlock() + log.L(bcm.ctx).Debugf("Checking block height confirmations for %d pending items headBlock=%d", len(items), headBlock) for _, p := range items { if err := bcm.confirmationCheckUsingHeadBlockNumber(p); err != nil { log.L(bcm.ctx).Errorf("Block height confirmation refresh failed for %s: %s", p.getKey(), err) @@ -877,7 +937,9 @@ func (bcm *blockConfirmationManager) dispatchBlockHeightConfirmations(pending *p confirmed := confirmationCount == bcm.requiredConfirmations if confirmed { + receiptValidationStartTime := time.Now() // do confirmation check here to ensure the transaction receipt is still valid + log.L(bcm.ctx).Debugf("Validating transaction receipt on confirmation listener item=%s", pending.getKey()) res, reason, err := bcm.connector.TransactionReceipt(bcm.ctx, &ffcapi.TransactionReceiptRequest{ TransactionHash: pending.transactionHash, }) @@ -888,6 +950,8 @@ func (bcm *blockConfirmationManager) dispatchBlockHeightConfirmations(pending *p pending.blockNumber = 0 pending.previousConfirmationCount = nil bcm.receiptChecker.schedule(pending, true) + } else { + log.L(bcm.ctx).Errorf("Confirmation listener receipt validation failed item=%s duration=%s: %s", pending.getKey(), time.Since(receiptValidationStartTime), err) } return err } diff --git a/internal/confirmations/confirmations_test.go b/internal/confirmations/confirmations_test.go index e4677ae9..d9d3f55e 100644 --- a/internal/confirmations/confirmations_test.go +++ b/internal/confirmations/confirmations_test.go @@ -1091,8 +1091,51 @@ func TestProcessBlockHashesLookupFail(t *testing.T) { bcm.processBlockHashes([]string{ blockHash, - }) + }, true) + + mca.AssertExpectations(t) +} + +// TestProcessBlockHashesLightModeDoesNotSweepOnNotificationOnlyTrigger is the regression test for +// the confirmation-manager stall under sustained 429s: in light chain-tracking mode, a loop +// iteration triggered only by a notification (e.g. a receiptArrived from the receipt-checker pool, +// or a new transaction being tracked) must not re-run the full confirmation sweep over every +// pending item - only an actual new block event should. Before the fix, this alone would call +// TransactionReceipt for every pending item on every notification, an O(N x M) cost that degrades +// into an unrecoverable backlog under load. +func TestProcessBlockHashesLightModeDoesNotSweepOnNotificationOnlyTrigger(t *testing.T) { + bcm, mca := newTestBlockConfirmationManagerHeadBlockNumber() + emm := &metricsmocks.EventMetricsEmitter{} + bcm.receiptChecker = newReceiptChecker(bcm, 0, emm) + + txHash := "0x531e219d98d81dc9f9a14811ac537479f5d77a74bdba47629bfbebe2d7663ce7" + blockHash := "0x0e32d749a86cfaf551d528b5b121cea456f980a39e5b8136eb8e85dbc744a542" + bcm.headBlockNumber = 1004 + pending := &pendingItem{ + pType: pendingTypeTransaction, + transactionHash: txHash, + blockHash: blockHash, + blockNumber: 1001, // 1004-1001 == the 3 confirmations required by newTestBlockConfirmationManagerHeadBlockNumber() + confirmationsCallback: func(ctx context.Context, notification *apitypes.ConfirmationsNotification) { + }, + } + bcm.pending[pending.getKey()] = pending + + bcm.processBlockHashes(nil, false /* notification-only trigger */) + mca.AssertNotCalled(t, "TransactionReceipt", mock.Anything, mock.Anything) + // A genuine new block event must still trigger the sweep - even though light mode block events + // carry no populated block hashes, only a head number bump (newBlockEvent=true is the correct + // signal, not len(blockHashes)). + mca.On("TransactionReceipt", mock.Anything, mock.MatchedBy(func(r *ffcapi.TransactionReceiptRequest) bool { + return r.TransactionHash == txHash + })).Return(&ffcapi.TransactionReceiptResponse{ + TransactionReceiptResponseBase: ffcapi.TransactionReceiptResponseBase{ + BlockNumber: fftypes.NewFFBigInt(1001), + BlockHash: blockHash, + }, + }, ffcapi.ErrorReason(""), nil).Once() + bcm.processBlockHashes(nil, true /* new block event */) mca.AssertExpectations(t) } @@ -1233,7 +1276,7 @@ func TestCheckReceiptImmediateConfirm(t *testing.T) { }, } blocks := bcm.newBlockState() - go bcm.dispatchReceipt(pending, receipt, blocks) + go bcm.dispatchReceipt(pending, receipt, 1, blocks) <-done } @@ -1263,7 +1306,47 @@ func TestCheckReceiptWalkFail(t *testing.T) { }, } blocks := bcm.newBlockState() - bcm.dispatchReceipt(pending, receipt, blocks) + bcm.dispatchReceipt(pending, receipt, 1, blocks) +} + +func TestDispatchReceiptIgnoresStaleGeneration(t *testing.T) { + + bcm, mca := newTestBlockConfirmationManager() + mca.On("BlockInfoByNumber", mock.Anything, mock.MatchedBy(func(r *ffcapi.BlockInfoByNumberRequest) bool { + return r.BlockNumber.Uint64() == 1002 + })).Return(nil, ffcapi.ErrorReasonNotFound, fmt.Errorf("not found")) + + forkA := &ffcapi.TransactionReceiptResponse{ + TransactionReceiptResponseBase: ffcapi.TransactionReceiptResponseBase{ + BlockNumber: fftypes.NewFFBigInt(1001), + BlockHash: "0xea681fadcf56ee6254a0d30b255c56636ee9199c73c45f0dd5823759b2ad1ef8", + }, + } + forkB := &ffcapi.TransactionReceiptResponse{ + TransactionReceiptResponseBase: ffcapi.TransactionReceiptResponseBase{ + BlockNumber: fftypes.NewFFBigInt(1001), + BlockHash: "0x33eb56730878a08e126f2d52b19242d3b3127dc7611447255928be91b2dda455", + }, + } + + txHash := "0x1dcc4de8dec75d7aab85b567b6ccd41ad312451b948a7413f0a142fd40d49347" + pending := &pendingItem{ + pType: pendingTypeTransaction, + transactionHash: txHash, + } + blocks := bcm.newBlockState() + + // Newer receipt applied first (as can happen when receiptArrived notifications arrive out of order). + bcm.dispatchReceipt(pending, forkB, 2, blocks) + assert.Equal(t, forkB.BlockHash, pending.blockHash) + assert.Equal(t, uint64(2), pending.appliedReceiptGeneration) + + // Older in-flight receipt must not overwrite the newer one. + bcm.dispatchReceipt(pending, forkA, 1, blocks) + assert.Equal(t, forkB.BlockHash, pending.blockHash) + assert.Equal(t, uint64(2), pending.appliedReceiptGeneration) + + mca.AssertExpectations(t) } func TestScheduleReceiptCheck(t *testing.T) { @@ -1299,6 +1382,67 @@ func TestScheduleReceiptCheck(t *testing.T) { } +// TestScheduleReceiptChecksLightModeRetriesUnreceiptedItemsEveryBlock is the regression test for +// why light mode must retry an outstanding receipt check on every new block: unlike full mode +// (which actively detects a mined transaction by scanning each new block's transaction list, see +// processBlock), light mode has no way to know which block will contain a given pending +// transaction. So an item that already had its first check (scheduledAtLeastOnce=true) but got +// "not found" - still no blockHash - must be retried on the very next block, not left to wait for +// the 60s stale-receipt-timeout. +func TestScheduleReceiptChecksLightModeRetriesUnreceiptedItemsEveryBlock(t *testing.T) { + + bcm, _ := newTestBlockConfirmationManagerHeadBlockNumber() // light mode + emm := &metricsmocks.EventMetricsEmitter{} + bcm.receiptChecker = newReceiptChecker(bcm, 0, emm) + + pendingNoReceiptYet := &pendingItem{ // already checked once, "not found" - must be retried + pType: pendingTypeTransaction, + lastReceiptCheck: time.Now(), // just checked - nowhere near the stale-timeout + transactionHash: "0x1dcc4de8dec75d7aab85b567b6ccd41ad312451b948a7413f0a142fd40d49347", + scheduledAtLeastOnce: true, + blockHash: "", + } + pendingAlreadyHasReceipt := &pendingItem{ // already has a receipt - not this path's concern + pType: pendingTypeTransaction, + lastReceiptCheck: time.Now(), + transactionHash: "0x531e219d98d81dc9f9a14811ac537479f5d77a74bdba47629bfbebe2d7663ce7", + scheduledAtLeastOnce: true, + blockHash: "0x0e32d749a86cfaf551d528b5b121cea456f980a39e5b8136eb8e85dbc744a542", + } + bcm.pending[pendingNoReceiptYet.getKey()] = pendingNoReceiptYet + bcm.pending[pendingAlreadyHasReceipt.getKey()] = pendingAlreadyHasReceipt + + bcm.scheduleReceiptChecks(true) // simulates a new light-mode block event + + assert.Equal(t, 1, bcm.receiptChecker.entries.Len()) + assert.Equal(t, pendingNoReceiptYet, bcm.receiptChecker.entries.Front().Value.(*pendingItem)) +} + +// TestScheduleReceiptChecksFullModeDoesNotRetryUnreceiptedItems is the guard test for the race we +// found and reverted: applying the light-mode retry-every-block behavior in full mode too would +// race against processBlock's own active scheduling of the same item (it caused a duplicate +// in-flight receipt check against TestBlockConfirmationManagerE2ETransactionMovedFork). Full mode +// must only ever schedule a not-yet-scheduled item, never re-trigger on blockHash=="" alone. +func TestScheduleReceiptChecksFullModeDoesNotRetryUnreceiptedItems(t *testing.T) { + + bcm, _ := newTestBlockConfirmationManager() // full mode + emm := &metricsmocks.EventMetricsEmitter{} + bcm.receiptChecker = newReceiptChecker(bcm, 0, emm) + + pendingAlreadyScheduledNoReceiptYet := &pendingItem{ + pType: pendingTypeTransaction, + lastReceiptCheck: time.Now(), // just checked - nowhere near the stale-timeout + transactionHash: "0x1dcc4de8dec75d7aab85b567b6ccd41ad312451b948a7413f0a142fd40d49347", + scheduledAtLeastOnce: true, + blockHash: "", + } + bcm.pending[pendingAlreadyScheduledNoReceiptYet.getKey()] = pendingAlreadyScheduledNoReceiptYet + + bcm.scheduleReceiptChecks(true) + + assert.Equal(t, 0, bcm.receiptChecker.entries.Len()) +} + func TestBlockState(t *testing.T) { bcm, mca := newTestBlockConfirmationManager() @@ -1826,3 +1970,68 @@ func TestBlockConfirmationManagerHeadBlockNumberNoOpWithoutReceipt(t *testing.T) bcm.Stop() mca.AssertExpectations(t) } + +// TestBlockConfirmationManagerLightModeChecksReceiptOnNextBlockNotStaleTimeout is the end-to-end +// regression test for the light-mode performance fix: a transaction added to an already-running +// light-mode manager (i.e. not the very first block the manager has ever seen - matching the real +// scenario where transactions arrive continuously over a long-running process) must have its +// receipt checked on the very next new block event, not have to wait for the default 60s +// stale-receipt-timeout. +func TestBlockConfirmationManagerLightModeChecksReceiptOnNextBlockNotStaleTimeout(t *testing.T) { + bcm, mca := newTestBlockConfirmationManagerHeadBlockNumber() + config.Set(tmconfig.ConfirmationsReceiptWorkers, 1) // need a real worker to consume the schedule + + txHash := "0x531e219d98d81dc9f9a14811ac537479f5d77a74bdba47629bfbebe2d7663ce7" + blockHash := "0x0e32d749a86cfaf551d528b5b121cea456f980a39e5b8136eb8e85dbc744a542" + + receiptChecked := make(chan struct{}, 1) + mca.On("TransactionReceipt", mock.Anything, mock.MatchedBy(func(r *ffcapi.TransactionReceiptRequest) bool { + return r.TransactionHash == txHash + })).Run(func(mock.Arguments) { + receiptChecked <- struct{}{} + }).Return(&ffcapi.TransactionReceiptResponse{ + TransactionReceiptResponseBase: ffcapi.TransactionReceiptResponseBase{ + BlockNumber: fftypes.NewFFBigInt(1001), + BlockHash: blockHash, + }, + }, ffcapi.ErrorReason(""), nil).Maybe() + + bcm.Start() + ch := bcm.GetReceiveChannel() + + // Establish the manager's head well before the transaction even exists - this is deliberately + // NOT "the first block ever", matching how a long-running process actually behaves. + ch <- &ffcapi.BlockHashEvent{HeadBlockNumber: 900} + + assert.NoError(t, bcm.Notify(&Notification{ + NotificationType: NewTransaction, + Transaction: &TransactionInfo{ + TransactionHash: txHash, + Receipt: func(ctx context.Context, receipt *ffcapi.TransactionReceiptResponse) {}, + Confirmations: func(ctx context.Context, notification *apitypes.ConfirmationsNotification) {}, + }, + })) + + // The NewTransaction notification and the block events travel over separate channels, so wait + // for it to actually land in bcm.pending before sending the block event that's supposed to + // trigger its receipt check - otherwise the two channels can race and the block event could be + // (and, roughly 1 in 5 test runs, was) consumed before the notification, making this test flaky + // for a reason that has nothing to do with the behavior under test. + pendingKey := pendingKeyForTX(txHash) + assert.Eventually(t, func() bool { + bcm.pendingMux.Lock() + defer bcm.pendingMux.Unlock() + return bcm.pending[pendingKey] != nil + }, time.Second, 5*time.Millisecond) + + // A single subsequent new block event is all it should take. + ch <- &ffcapi.BlockHashEvent{HeadBlockNumber: 901} + + select { + case <-receiptChecked: + case <-time.After(time.Second): + t.Fatal("timeout waiting for TransactionReceipt to be checked - should not need to wait for the stale-receipt-timeout") + } + + bcm.Stop() +} diff --git a/internal/confirmations/receipt_checker.go b/internal/confirmations/receipt_checker.go index 7d2fec26..ec936ddc 100644 --- a/internal/confirmations/receipt_checker.go +++ b/internal/confirmations/receipt_checker.go @@ -43,7 +43,7 @@ type receiptChecker struct { cond *sync.Cond entries *list.List metricsEmitter metrics.ReceiptCheckerMetricsEmitter - notify func(*pendingItem, *ffcapi.TransactionReceiptResponse) + notify func(*pendingItem, *ffcapi.TransactionReceiptResponse, uint64) } func newReceiptChecker(bcm *blockConfirmationManager, workerCount int, rcme metrics.ReceiptCheckerMetricsEmitter) *receiptChecker { @@ -52,11 +52,12 @@ func newReceiptChecker(bcm *blockConfirmationManager, workerCount int, rcme metr workerCount: workerCount, workersDone: make([]chan struct{}, workerCount), metricsEmitter: rcme, - notify: func(pending *pendingItem, receipt *ffcapi.TransactionReceiptResponse) { + notify: func(pending *pendingItem, receipt *ffcapi.TransactionReceiptResponse, receiptGeneration uint64) { _ = bcm.Notify(&Notification{ - NotificationType: receiptArrived, - pending: pending, - receipt: receipt, + NotificationType: receiptArrived, + pending: pending, + receipt: receipt, + receiptGeneration: receiptGeneration, }) }, } @@ -100,6 +101,9 @@ func (rc *receiptChecker) run(i int) { if pending == nil { return false /* exit the retry loop with err */, i18n.NewError(ctx, tmmsgs.MsgShuttingDown) } + rc.cond.L.Lock() + checkGeneration := pending.receiptGeneration + rc.cond.L.Unlock() res, reason, receiptErr := rc.bcm.connector.TransactionReceipt(ctx, &ffcapi.TransactionReceiptRequest{ TransactionHash: pending.transactionHash, @@ -127,7 +131,7 @@ func (rc *receiptChecker) run(i int) { // Dispatch the receipt back to the main routine. if res != nil { rc.metricsEmitter.RecordReceiptCheckMetrics(ctx, "notified", time.Since(startTime).Seconds()) - rc.notify(pending, res) + rc.notify(pending, res, checkGeneration) } else { rc.metricsEmitter.RecordReceiptCheckMetrics(ctx, "empty", time.Since(startTime).Seconds()) } @@ -148,6 +152,7 @@ func (rc *receiptChecker) schedule(pending *pendingItem, suspectedTimeout bool) rc.cond.L.Unlock() return } + pending.receiptGeneration++ pending.queuedStale = rc.entries.PushBack(pending) pending.scheduledAtLeastOnce = true rc.cond.Signal()