From 6cf7e08f4acff0af9dbe37cc7523fc7e5d3a2f4b Mon Sep 17 00:00:00 2001 From: Dirk Brink Date: Mon, 29 Dec 2025 12:56:38 -0800 Subject: [PATCH 1/6] node: Accountant audit improvements --- node/pkg/accountant/accountant.go | 86 ++-- node/pkg/accountant/accountant_test.go | 522 ++++++++++++++++++++++++- node/pkg/accountant/audit.go | 392 ++++++++++++------- node/pkg/accountant/data_for_test.go | 123 ++++++ node/pkg/accountant/metrics.go | 5 + node/pkg/accountant/query_test.go | 260 ++++++++++-- node/pkg/accountant/submit_obs.go | 4 +- node/pkg/accountant/watcher.go | 5 +- node/pkg/wormconn/send_tx.go | 2 +- 9 files changed, 1196 insertions(+), 203 deletions(-) diff --git a/node/pkg/accountant/accountant.go b/node/pkg/accountant/accountant.go index 8f6563fdac7..48aa09e4aa8 100644 --- a/node/pkg/accountant/accountant.go +++ b/node/pkg/accountant/accountant.go @@ -101,7 +101,10 @@ type Accountant struct { } // On startup, there can be a large number of re-submission requests. -const subChanSize = 500 +const subChanSize = 5000 + +// auditSubmitTimeout is the timeout for blocking channel writes during audit. +const auditSubmitTimeout = 30 * time.Second // baseEnabled returns true if the base accountant is enabled, false if not. func (acct *Accountant) baseEnabled() bool { @@ -315,7 +318,11 @@ func (acct *Accountant) SubmitObservation(msg *common.MessagePublication) (bool, acct.pendingTransfersLock.Lock() defer acct.pendingTransfersLock.Unlock() - // If this is already pending, don't send it again. + var pe *pendingEntry + + // If there is a digest mismatch, don't send it again. + // Otherwise resubmit it and rely on the submitPending flag to prevent duplicate submissions to the contract. + // This allows manual reobservations to proceed. if oldEntry, exists := acct.pendingTransfers[msgId]; exists { if oldEntry.digest != digest { digestMismatches.Inc() @@ -325,17 +332,18 @@ func (acct *Accountant) SubmitObservation(msg *common.MessagePublication) (bool, zap.String("newDigest", digest), zap.Bool("enforcing", enforceFlag), ) - } else { - acct.logger.Info("blocking transfer because it is already outstanding", zap.String("msgID", msgId), zap.Bool("enforcing", enforceFlag)) - } - return !enforceFlag, nil - } - // Add it to the pending map and the database. - pe := &pendingEntry{msg: msg, msgId: msgId, digest: digest, isNTT: isNTT, enforceFlag: enforceFlag} - if err := acct.addPendingTransferAlreadyLocked(pe); err != nil { - acct.logger.Error("failed to persist pending transfer, blocking publishing", zap.String("msgID", msgId), zap.Error(err)) - return false, err + return !enforceFlag, nil + } + pe = oldEntry + } else { + // Add it to the pending map and the database. + // We only add it if it is not already present. + pe = &pendingEntry{msg: msg, msgId: msgId, digest: digest, isNTT: isNTT, enforceFlag: enforceFlag} + if err := acct.addPendingTransferAlreadyLocked(pe); err != nil { + acct.logger.Error("failed to persist pending transfer, blocking publishing", zap.String("msgID", msgId), zap.Error(err)) + return false, err + } } // This transaction may take a while. Pass it off to the worker so we don't block the processor. @@ -345,7 +353,7 @@ func (acct *Accountant) SubmitObservation(msg *common.MessagePublication) (bool, tag = "ntt-accountant" } acct.logger.Info(fmt.Sprintf("submitting transfer to %s for approval", tag), zap.String("msgID", msgId), zap.Bool("canPublish", !enforceFlag)) - _ = acct.submitObservation(pe) + _ = acct.submitObservation(acct.ctx, pe, false) // Non-blocking from processor } // If we are not enforcing accountant, the event can be published. Otherwise we have to wait to hear back from the contract. @@ -436,9 +444,10 @@ func (acct *Accountant) loadPendingTransfers() error { // submitObservation sends an observation request to the worker so it can be submitted to the contract. If the transfer is already // marked as "submit pending", this function returns false without doing anything. Otherwise it returns true. The return value can -// be used to avoid unnecessary error logging. If writing to the channel would block, this function returns without doing anything, -// assuming the pending transfer will be handled on the next audit interval. This function grabs the state lock. -func (acct *Accountant) submitObservation(pe *pendingEntry) bool { +// be used to avoid unnecessary error logging. If blocking is false and writing to the channel would block, this function returns +// without doing anything, assuming the pending transfer will be handled on the next audit interval. If blocking is true, it will +// block until the channel has space, a timeout occurs, or the context is cancelled. This function grabs the state lock. +func (acct *Accountant) submitObservation(ctx context.Context, pe *pendingEntry, blocking bool) bool { pe.stateLock.Lock() defer pe.stateLock.Unlock() @@ -449,24 +458,47 @@ func (acct *Accountant) submitObservation(pe *pendingEntry) bool { pe.state.submitPending = true pe.state.updTime = time.Now() + timeout := time.Duration(0) + if blocking { + timeout = auditSubmitTimeout + } + if pe.isNTT { - acct.submitToChannel(pe, acct.nttSubChan, "ntt-accountant") + acct.submitToChannel(ctx, pe, acct.nttSubChan, "ntt-accountant", blocking, timeout) } else { - acct.submitToChannel(pe, acct.subChan, "accountant") + acct.submitToChannel(ctx, pe, acct.subChan, "accountant", blocking, timeout) } return true } -// submitToChannel submits an observation to the specified channel. If the submission fails because the channel is full, -// it marks the transfer as pending so it will be resubmitted by the audit. -func (acct *Accountant) submitToChannel(pe *pendingEntry, subChan chan *common.MessagePublication, tag string) { - select { - case subChan <- pe.msg: - acct.logger.Debug(fmt.Sprintf("submitted observation to channel for %s", tag), zap.String("msgId", pe.msgId)) - default: - acct.logger.Error(fmt.Sprintf("unable to submit observation to %s because the channel is full, will try next interval", tag), zap.String("msgId", pe.msgId)) - pe.state.submitPending = false +// submitToChannel submits an observation to the specified channel. If blocking is false and the channel is full, +// it marks the transfer as no longer pending so it will be resubmitted by the audit. If blocking is true, it will +// block until the channel has space, a timeout occurs, or the context is cancelled. +func (acct *Accountant) submitToChannel(ctx context.Context, pe *pendingEntry, subChan chan *common.MessagePublication, tag string, blocking bool, timeout time.Duration) { + if blocking { + select { + case subChan <- pe.msg: + acct.logger.Debug(fmt.Sprintf("submitted observation to channel for %s", tag), zap.String("msgId", pe.msgId)) + case <-time.After(timeout): + channelSubmitTimeouts.Inc() + acct.logger.Warn(fmt.Sprintf("timeout submitting observation to %s channel, will retry next audit", tag), + zap.String("msgId", pe.msgId), + zap.Duration("timeout", timeout)) + pe.state.submitPending = false + case <-ctx.Done(): + acct.logger.Debug(fmt.Sprintf("context cancelled while submitting to %s channel", tag), zap.String("msgId", pe.msgId)) + pe.state.submitPending = false + } + } else { + // Non-blocking write (existing behavior for processor) + select { + case subChan <- pe.msg: + acct.logger.Debug(fmt.Sprintf("submitted observation to channel for %s", tag), zap.String("msgId", pe.msgId)) + default: + acct.logger.Error(fmt.Sprintf("unable to submit observation to %s because the channel is full, will try next interval", tag), zap.String("msgId", pe.msgId)) + pe.state.submitPending = false + } } } diff --git a/node/pkg/accountant/accountant_test.go b/node/pkg/accountant/accountant_test.go index f5d5ab11471..86a688153c8 100644 --- a/node/pkg/accountant/accountant_test.go +++ b/node/pkg/accountant/accountant_test.go @@ -3,8 +3,11 @@ package accountant import ( "context" "encoding/binary" + "encoding/hex" "encoding/json" + "errors" "math/big" + "strings" "sync" "testing" "time" @@ -90,6 +93,28 @@ func (c *MockAccountantWormchainConn) WaitUntilTxRespConsumed() { } } +// AuditMockWormchainConn extends MockAccountantWormchainConn with configurable +// query responses for audit testing. +type AuditMockWormchainConn struct { + MockAccountantWormchainConn + + allPendingTransfersResp []byte + allPendingTransfersErr error + batchTransferStatusResp []byte + batchTransferStatusErr error +} + +func (c *AuditMockWormchainConn) SubmitQuery(ctx context.Context, contractAddress string, query []byte) ([]byte, error) { + queryStr := string(query) + if strings.Contains(queryStr, "all_pending_transfers") { + return c.allPendingTransfersResp, c.allPendingTransfersErr + } + if strings.Contains(queryStr, "batch_transfer_status") { + return c.batchTransferStatusResp, c.batchTransferStatusErr + } + return []byte{}, nil +} + func newAccountantForTest( t *testing.T, logger *zap.Logger, @@ -147,12 +172,12 @@ func hashToTxID(str string) []byte { // Note this method assumes 18 decimals for the amount. func buildMockTransferPayloadBytes( - t uint8, - tokenChainID vaa.ChainID, - tokenAddrStr string, - toChainID vaa.ChainID, - toAddrStr string, - amtFloat float64, + t uint8, //nolint:unparam + tokenChainID vaa.ChainID, //nolint:unparam + tokenAddrStr string, //nolint:unparam + toChainID vaa.ChainID, //nolint:unparam + toAddrStr string, //nolint:unparam + amtFloat float64, //nolint:unparam ) []byte { bytes := make([]byte, 101) bytes[0] = t @@ -654,3 +679,488 @@ func TestCreateAuditMapFiltersNTT(t *testing.T) { assert.Equal(t, 1, len(entries)) assert.Equal(t, pe2.msgId, entries[0].msgId) } + +// createTestPendingEntry creates a pendingEntry for testing with semi realistic values. +// The digest is computed from the message using CreateDigest(). +func createTestPendingEntry( + emitterChain vaa.ChainID, //nolint:unparam + emitterAddr vaa.Address, + sequence uint64, + txHash []byte, + payload []byte, +) *pendingEntry { + msg := &common.MessagePublication{ + TxID: txHash, + Timestamp: time.Unix(1654543099, 0), + Nonce: 1, + Sequence: sequence, + EmitterChain: emitterChain, + EmitterAddress: emitterAddr, + ConsistencyLevel: 32, + Payload: payload, + } + return &pendingEntry{ + msg: msg, + msgId: msg.MessageIDString(), + digest: msg.CreateDigest(), + enforceFlag: true, + } +} + +// drainObsvReqChannel drains the observation request channel +func drainObsvReqChannel(ch chan *gossipv1.ObservationRequest) { + for { + select { + case <-ch: + default: + return + } + } +} + +// drainMsgChannel drains a message publication channel +func drainMsgChannel(ch chan *common.MessagePublication) { + for { + select { + case <-ch: + default: + return + } + } +} + +// newAccountantForAuditTest creates an accountant configured for audit testing +// with an AuditMockWormchainConn. +func newAccountantForAuditTest( + t *testing.T, + logger *zap.Logger, + ctx context.Context, + obsvReqWriteC chan *gossipv1.ObservationRequest, + acctWriteC chan *common.MessagePublication, + wormchainConn *AuditMockWormchainConn, +) *Accountant { + var db guardianDB.MockAccountantDB + + pk := devnet.InsecureDeterministicEcdsaKeyByIndex(uint64(0)) + guardianSigner, err := guardiansigner.GenerateSignerWithPrivatekeyUnsafe(pk) + require.NoError(t, err) + + gst := common.NewGuardianSetState(nil) + // Guardian set with index 0, our guardian at index 0 + gs := &common.GuardianSet{ + Index: 0, + Keys: []ethCommon.Address{ethCommon.HexToAddress("0xbeFA429d57cD18b7F8A4d91A2da9AB4AF05d0FBe")}, + } + gst.Set(gs) + + acct := NewAccountant( + ctx, + logger, + &db, + obsvReqWriteC, + "0xdeadbeef", + "none", + wormchainConn, + true, // enforceFlag + "", + nil, + guardianSigner, + gst, + acctWriteC, + common.GoTest, // Use GoTest to avoid starting worker goroutines that consume from subChan + ) + + err = acct.Start(ctx) + require.NoError(t, err) + return acct +} + +// Standard test values for audit tests +var ( + testEmitterAddr, _ = vaa.StringToAddress("0000000000000000000000000290fb167208af455bb137780163b7b7a9a10c16") + testEmitterAddrStr = "0000000000000000000000000290fb167208af455bb137780163b7b7a9a10c16" + testSequence = uint64(1674568234) + testTxHash = []byte{0x06, 0xf5, 0x41, 0xf5, 0xec, 0xfc, 0x43, 0x40, 0x7c, 0x31, 0x58, 0x7a, 0xa6, 0xac, 0x3a, 0x68, 0x9e, 0x89, 0x60, 0xf3, 0x6d, 0xc2, 0x3c, 0x33, 0x2d, 0xb5, 0x51, 0x0d, 0xfc, 0x6a, 0x40, 0x64} +) + +func TestPerformAuditResubmitsUnsignedTransfer(t *testing.T) { + ctx := context.Background() + logger := zaptest.NewLogger(t) + obsvReqWriteC := make(chan *gossipv1.ObservationRequest, 10) + acctChan := make(chan *common.MessagePublication, MsgChannelCapacity) + + payload := buildMockTransferPayloadBytes(1, vaa.ChainIDEthereum, "0x707f9118e33a9b8998bea41dd0d46f38bb963fc8", vaa.ChainIDPolygon, "0x707f9118e33a9b8998bea41dd0d46f38bb963fc8", 1.25) + + // Create pending entry + pe := createTestPendingEntry(vaa.ChainIDEthereum, testEmitterAddr, testSequence, testTxHash, payload) + + // Decode the digest from hex to bytes for the mock response + digestBytes, err := hex.DecodeString(pe.digest) + require.NoError(t, err) + + // Mock returns pending transfer where guardian 0 hasn't signed (signatures="0") + allPendingResp := createPendingTransfersForTestWithTxHash( + uint16(vaa.ChainIDEthereum), + testEmitterAddrStr, + testSequence, + testTxHash, + digestBytes, + 0, // guardian set index + "0", // signatures - guardian 0 has NOT signed + ) + + wormchainConn := &AuditMockWormchainConn{ + allPendingTransfersResp: allPendingResp, + } + + acct := newAccountantForAuditTest(t, logger, ctx, obsvReqWriteC, acctChan, wormchainConn) + + // Add the pending entry to the accountant's map + acct.pendingTransfersLock.Lock() + acct.pendingTransfers[pe.msgId] = pe + acct.pendingTransfersLock.Unlock() + + // Create tmpMap for audit (simulating what createAuditMap does) + tmpMap := map[string][]*pendingEntry{ + pe.makeAuditKey(): {pe}, + } + + // Run the audit + acct.performAudit(ctx, tmpMap, wormchainConn, "test-contract") + + // Verify: entry should have been submitted to subChan + assert.Equal(t, 1, len(acct.subChan), "expected 1 message in subChan") + + // Verify: entry should have been removed from tmpMap because phase 1 found it + // in the contract's pending list and we hadn't signed it, so it was deleted + // from tmpMap after being resubmitted. + assert.Equal(t, 0, len(tmpMap), "expected tmpMap to be empty") + + // Verify: submitPending flag should be set + assert.True(t, pe.submitPending(), "expected submitPending to be true") + + // Drain channels + drainMsgChannel(acct.subChan) + drainObsvReqChannel(obsvReqWriteC) +} + +func TestPerformAuditRequestsReobservation(t *testing.T) { + ctx := context.Background() + logger := zaptest.NewLogger(t) + obsvReqWriteC := make(chan *gossipv1.ObservationRequest, 10) + acctChan := make(chan *common.MessagePublication, MsgChannelCapacity) + + // Mock returns pending transfer where guardian 0 hasn't signed + // but we DON'T have this transfer locally + allPendingResp := createPendingTransfersForTestWithTxHash( + uint16(vaa.ChainIDEthereum), + testEmitterAddrStr, + testSequence, + testTxHash, + []byte("somedigest"), + 0, // guardian set index + "0", // signatures - guardian 0 has NOT signed + ) + + wormchainConn := &AuditMockWormchainConn{ + allPendingTransfersResp: allPendingResp, + } + + acct := newAccountantForAuditTest(t, logger, ctx, obsvReqWriteC, acctChan, wormchainConn) + + // Empty local map - we don't have this transfer + tmpMap := map[string][]*pendingEntry{} + + // Run the audit + acct.performAudit(ctx, tmpMap, wormchainConn, "test-contract") + + // Verify: reobservation request should have been sent + assert.Equal(t, 1, len(obsvReqWriteC), "expected 1 reobservation request") + + // Verify the request has correct values + req := <-obsvReqWriteC + assert.Equal(t, uint32(vaa.ChainIDEthereum), req.ChainId) + assert.Equal(t, testTxHash, req.TxHash) + + // Verify: tmpMap remains empty because we started with an empty local map. + // The contract's pending transfer was handled via reobservation request, not + // via tmpMap lookup, so tmpMap was never modified. + assert.Equal(t, 0, len(tmpMap), "expected tmpMap to remain empty") + + // Drain channels + drainObsvReqChannel(obsvReqWriteC) +} + +func TestPerformAuditSkipsAlreadySigned(t *testing.T) { + ctx := context.Background() + logger := zaptest.NewLogger(t) + obsvReqWriteC := make(chan *gossipv1.ObservationRequest, 10) + acctChan := make(chan *common.MessagePublication, MsgChannelCapacity) + + payload := buildMockTransferPayloadBytes(1, vaa.ChainIDEthereum, "0x707f9118e33a9b8998bea41dd0d46f38bb963fc8", vaa.ChainIDPolygon, "0x707f9118e33a9b8998bea41dd0d46f38bb963fc8", 1.25) + pe := createTestPendingEntry(vaa.ChainIDEthereum, testEmitterAddr, testSequence, testTxHash, payload) + digestBytes, err := hex.DecodeString(pe.digest) + require.NoError(t, err) + + // Mock returns pending transfer where guardian 0 HAS signed (signatures="1") + allPendingResp := createPendingTransfersForTestWithTxHash( + uint16(vaa.ChainIDEthereum), + testEmitterAddrStr, + testSequence, + testTxHash, + digestBytes, + 0, // guardian set index + "1", // signatures - guardian 0 HAS signed (bit 0 set) + ) + + wormchainConn := &AuditMockWormchainConn{ + allPendingTransfersResp: allPendingResp, + } + + acct := newAccountantForAuditTest(t, logger, ctx, obsvReqWriteC, acctChan, wormchainConn) + + acct.pendingTransfersLock.Lock() + acct.pendingTransfers[pe.msgId] = pe + acct.pendingTransfersLock.Unlock() + + tmpMap := map[string][]*pendingEntry{ + pe.makeAuditKey(): {pe}, + } + + // Run the audit + acct.performAudit(ctx, tmpMap, wormchainConn, "test-contract") + + // Verify: nothing should have been submitted (we already signed) + assert.Equal(t, 0, len(acct.subChan), "expected no messages in subChan") + assert.Equal(t, 0, len(obsvReqWriteC), "expected no reobservation requests") + + // Verify: tmpMap still contains the entry because phase 1 skipped it + // (guardian already signed), so delete(tmpMap) was never called. + // Phase 2 iterates remaining tmpMap entries but never removes them from tmpMap. + assert.Equal(t, 1, len(tmpMap), "expected tmpMap to still contain the entry") + + // Drain channels + drainMsgChannel(acct.subChan) + drainObsvReqChannel(obsvReqWriteC) +} + +func TestPerformAuditPhase2Committed(t *testing.T) { + ctx := context.Background() + logger := zaptest.NewLogger(t) + obsvReqWriteC := make(chan *gossipv1.ObservationRequest, 10) + acctChan := make(chan *common.MessagePublication, MsgChannelCapacity) + + payload := buildMockTransferPayloadBytes(1, vaa.ChainIDEthereum, "0x707f9118e33a9b8998bea41dd0d46f38bb963fc8", vaa.ChainIDPolygon, "0x707f9118e33a9b8998bea41dd0d46f38bb963fc8", 1.25) + pe := createTestPendingEntry(vaa.ChainIDEthereum, testEmitterAddr, testSequence, testTxHash, payload) + + // Decode the digest from hex to bytes for the mock response + digestBytes, err := hex.DecodeString(pe.digest) + require.NoError(t, err) + + // Mock: all_pending_transfers returns empty (transfer not in pending list) + // Mock: batch_transfer_status returns committed with matching digest + batchStatusResp := createBatchTransferStatusResponse( + uint16(vaa.ChainIDEthereum), + testEmitterAddrStr, + testSequence, + "committed", + digestBytes, + ) + + wormchainConn := &AuditMockWormchainConn{ + allPendingTransfersResp: []byte(`{"pending":[]}`), + batchTransferStatusResp: batchStatusResp, + } + + acct := newAccountantForAuditTest(t, logger, ctx, obsvReqWriteC, acctChan, wormchainConn) + + acct.pendingTransfersLock.Lock() + acct.pendingTransfers[pe.msgId] = pe + acct.pendingTransfersLock.Unlock() + + tmpMap := map[string][]*pendingEntry{ + pe.makeAuditKey(): {pe}, + } + + // Run the audit + acct.performAudit(ctx, tmpMap, wormchainConn, "test-contract") + + // Verify: transfer should have been published to msgChan + assert.Equal(t, 1, len(acctChan), "expected 1 message in msgChan") + + // Verify: entry should have been removed from pendingTransfers + acct.pendingTransfersLock.Lock() + _, exists := acct.pendingTransfers[pe.msgId] + acct.pendingTransfersLock.Unlock() + assert.False(t, exists, "expected entry to be removed from pendingTransfers") + + // Verify: tmpMap still contains the entry because phase 2 never deletes from tmpMap. + // The entry was not in the contract's pending list (phase 1 had nothing to process), + // and phase 2 only reads tmpMap to query batch status — it publishes the committed + // transfer but does not remove it from tmpMap. + assert.Equal(t, 1, len(tmpMap), "expected tmpMap to still contain the entry") + + // Drain channels + drainMsgChannel(acctChan) + drainMsgChannel(acct.subChan) + drainObsvReqChannel(obsvReqWriteC) +} + +func TestPerformAuditPhase2Unknown(t *testing.T) { + ctx := context.Background() + logger := zaptest.NewLogger(t) + obsvReqWriteC := make(chan *gossipv1.ObservationRequest, 10) + acctChan := make(chan *common.MessagePublication, MsgChannelCapacity) + + payload := buildMockTransferPayloadBytes(1, vaa.ChainIDEthereum, "0x707f9118e33a9b8998bea41dd0d46f38bb963fc8", vaa.ChainIDPolygon, "0x707f9118e33a9b8998bea41dd0d46f38bb963fc8", 1.25) + pe := createTestPendingEntry(vaa.ChainIDEthereum, testEmitterAddr, testSequence, testTxHash, payload) + + // Mock: all_pending_transfers returns empty + // Mock: batch_transfer_status returns null (contract doesn't know about it) + batchStatusResp := createBatchTransferStatusResponse( + uint16(vaa.ChainIDEthereum), + testEmitterAddrStr, + testSequence, + "null", + nil, + ) + + wormchainConn := &AuditMockWormchainConn{ + allPendingTransfersResp: []byte(`{"pending":[]}`), + batchTransferStatusResp: batchStatusResp, + } + + acct := newAccountantForAuditTest(t, logger, ctx, obsvReqWriteC, acctChan, wormchainConn) + + acct.pendingTransfersLock.Lock() + acct.pendingTransfers[pe.msgId] = pe + acct.pendingTransfersLock.Unlock() + + tmpMap := map[string][]*pendingEntry{ + pe.makeAuditKey(): {pe}, + } + + // Run the audit + acct.performAudit(ctx, tmpMap, wormchainConn, "test-contract") + + // Verify: entry should have been resubmitted to subChan + assert.Equal(t, 1, len(acct.subChan), "expected 1 message in subChan") + + // Verify: submitPending flag should be set + assert.True(t, pe.submitPending(), "expected submitPending to be true") + + // Verify: tmpMap still contains the entry because phase 2 never deletes from tmpMap. + // The contract returned null status (unknown), so the entry was resubmitted, but + // tmpMap itself is only modified by delete() in phase 1. + assert.Equal(t, 1, len(tmpMap), "expected tmpMap to still contain the entry") + + // Drain channels + drainMsgChannel(acct.subChan) + drainObsvReqChannel(obsvReqWriteC) +} + +func TestPerformAuditPhase2DigestMismatch(t *testing.T) { + ctx := context.Background() + logger := zaptest.NewLogger(t) + obsvReqWriteC := make(chan *gossipv1.ObservationRequest, 10) + acctChan := make(chan *common.MessagePublication, MsgChannelCapacity) + + payload := buildMockTransferPayloadBytes(1, vaa.ChainIDEthereum, "0x707f9118e33a9b8998bea41dd0d46f38bb963fc8", vaa.ChainIDPolygon, "0x707f9118e33a9b8998bea41dd0d46f38bb963fc8", 1.25) + pe := createTestPendingEntry(vaa.ChainIDEthereum, testEmitterAddr, testSequence, testTxHash, payload) + + // Use a DIFFERENT digest than what the pending entry has + wrongDigestBytes := []byte("this_is_a_completely_different_digest_value_32b") + + // Mock: all_pending_transfers returns empty + // Mock: batch_transfer_status returns committed with DIFFERENT digest + batchStatusResp := createBatchTransferStatusResponse( + uint16(vaa.ChainIDEthereum), + testEmitterAddrStr, + testSequence, + "committed", + wrongDigestBytes, + ) + + wormchainConn := &AuditMockWormchainConn{ + allPendingTransfersResp: []byte(`{"pending":[]}`), + batchTransferStatusResp: batchStatusResp, + } + + acct := newAccountantForAuditTest(t, logger, ctx, obsvReqWriteC, acctChan, wormchainConn) + + acct.pendingTransfersLock.Lock() + acct.pendingTransfers[pe.msgId] = pe + acct.pendingTransfersLock.Unlock() + + tmpMap := map[string][]*pendingEntry{ + pe.makeAuditKey(): {pe}, + } + + // Run the audit + acct.performAudit(ctx, tmpMap, wormchainConn, "test-contract") + + // Verify: nothing should be published to msgChan (digest mismatch) + assert.Equal(t, 0, len(acctChan), "expected no messages in msgChan") + + // Verify: entry should have been removed from pendingTransfers (dropped) + acct.pendingTransfersLock.Lock() + _, exists := acct.pendingTransfers[pe.msgId] + acct.pendingTransfersLock.Unlock() + assert.False(t, exists, "expected entry to be removed from pendingTransfers") + + // Verify: tmpMap still contains the entry because phase 2 never deletes from tmpMap. + // The transfer was dropped from pendingTransfers due to digest mismatch, but + // tmpMap itself is only modified by delete() in phase 1. + assert.Equal(t, 1, len(tmpMap), "expected tmpMap to still contain the entry") + + // Drain channels + drainMsgChannel(acctChan) + drainMsgChannel(acct.subChan) + drainObsvReqChannel(obsvReqWriteC) +} + +func TestPerformAuditQueryError(t *testing.T) { + ctx := context.Background() + logger := zaptest.NewLogger(t) + obsvReqWriteC := make(chan *gossipv1.ObservationRequest, 10) + acctChan := make(chan *common.MessagePublication, MsgChannelCapacity) + + payload := buildMockTransferPayloadBytes(1, vaa.ChainIDEthereum, "0x707f9118e33a9b8998bea41dd0d46f38bb963fc8", vaa.ChainIDPolygon, "0x707f9118e33a9b8998bea41dd0d46f38bb963fc8", 1.25) + pe := createTestPendingEntry(vaa.ChainIDEthereum, testEmitterAddr, testSequence, testTxHash, payload) + + // Mock: all_pending_transfers returns error + wormchainConn := &AuditMockWormchainConn{ + allPendingTransfersErr: errors.New("query failed"), + } + + acct := newAccountantForAuditTest(t, logger, ctx, obsvReqWriteC, acctChan, wormchainConn) + + acct.pendingTransfersLock.Lock() + acct.pendingTransfers[pe.msgId] = pe + acct.pendingTransfersLock.Unlock() + + tmpMap := map[string][]*pendingEntry{ + pe.makeAuditKey(): {pe}, + } + + // Run the audit - should not panic + acct.performAudit(ctx, tmpMap, wormchainConn, "test-contract") + + // Verify: entry should still be in pendingTransfers (not removed due to error) + acct.pendingTransfersLock.Lock() + _, exists := acct.pendingTransfers[pe.msgId] + acct.pendingTransfersLock.Unlock() + assert.True(t, exists, "expected entry to remain in pendingTransfers") + + // Verify: nothing should have been sent to channels + assert.Equal(t, 0, len(acct.subChan), "expected no messages in subChan") + assert.Equal(t, 0, len(obsvReqWriteC), "expected no reobservation requests") + + // Verify: tmpMap still contains the entry because the query failed and + // performAudit returned early, so neither phase 1 nor phase 2 processed it. + assert.Equal(t, 1, len(tmpMap), "expected tmpMap to still contain the entry") + + // Drain channels + drainMsgChannel(acct.subChan) + drainObsvReqChannel(obsvReqWriteC) +} diff --git a/node/pkg/accountant/audit.go b/node/pkg/accountant/audit.go index a362e7d90b2..5d37939017e 100644 --- a/node/pkg/accountant/audit.go +++ b/node/pkg/accountant/audit.go @@ -1,15 +1,17 @@ // This code audits the set of pending transfers against the state reported by the smart contract. It has a runnable that is started when the accountant initializes. // It uses a ticker to periodically run the audit. The audit occurs in two phases that operate off of a temporary map of all pending transfers known to this guardian. // -// The first phase involves querying the smart contract for any observations that it thinks are missing for this guardian. The audit processes everything in the -// returned results and does one of the following: -// - If the observation is in our temporary map, we resubmit an observation to the contract and delete it from our temporary map. -// - If the observation is not in the temporary map, we request a reobservation from the local watcher. +// The first phase involves querying the smart contract for all pending transfers using the "all_pending_transfers" query. For each pending transfer returned: +// - If the transfer is for a different guardian set, it is skipped. +// - If this guardian has already signed (based on the signatures bitmask), it is skipped. +// - If the transfer is in our local pending map, we resubmit our observation to the contract. +// - If the transfer is not in our local map, we request a reobservation from the local watcher. // -// The second phase consists of requesting the status from the contract for everything that is still in the temporary map. For each returned item, we do the following: -// - If the contract indicates that the transfer has been committed, we validate the digest, then publish the transfer and delete it from the map. -// - If the contract indicates that the transfer is pending, we continue to wait for it to be committed. -// - If the contract indicates any other status (most likely meaning it does not know about it), we resubmit an observation to the contract. +// The second phase handles transfers that are in our local map but were NOT found in the contract's pending list. For each such transfer, we query +// the contract for its status using "batch_transfer_status": +// - If the contract indicates the transfer has been committed, we validate the digest and publish the transfer. +// - If the contract indicates the transfer is pending, we continue to wait. +// - If the contract indicates any other status (or doesn't know about it), we resubmit our observation. // // Note that any time we are considering resubmitting an observation to the contract, we first check the "submit pending" flag. If that is set, we do not // submit the observation to the contract, but continue to wait for it to work its way through the queue. @@ -21,6 +23,7 @@ import ( "encoding/hex" "encoding/json" "fmt" + "math/big" "strings" "time" @@ -42,15 +45,14 @@ const ( // maxPendingsPerQuery is the maximum number of pending transfers to submit in a single batch_transfer_status query to avoid gas errors. maxPendingsPerQuery = 500 + + // allPendingTransfersPageSize is the page size for all_pending_transfers query. + allPendingTransfersPageSize = 500 ) type ( - // MissingObservationsResponse is the result returned from the "missing_observations" query. - MissingObservationsResponse struct { - Missing []MissingObservation `json:"missing"` - } - - // MissingObservation is what is returned for a single missing observation. + // MissingObservation represents a single observation that the contract expects but we haven't submitted. + // Used when requesting reobservation from the local watcher. MissingObservation struct { ChainId uint16 `json:"chain_id"` TxHash []byte `json:"tx_hash"` @@ -95,22 +97,48 @@ type ( GuardianSetIndex uint32 `json:"guardian_set_index"` EmitterChain uint16 `json:"emitter_chain"` } + + // AllPendingTransfersResponse is the result from "all_pending_transfers" query. + AllPendingTransfersResponse struct { + Pending []PendingTransfer `json:"pending"` + } + + // PendingTransfer represents a single pending transfer from the contract. + PendingTransfer struct { + Key TransferKey `json:"key"` + Data []PendingTransferData `json:"data"` + } + + // PendingTransferData contains observation data for a pending transfer. + PendingTransferData struct { + Digest []byte `json:"digest"` + TxHash []byte `json:"tx_hash"` + Signatures string `json:"signatures"` // u128 as decimal string + GuardianSetIndex uint32 `json:"guardian_set_index"` + EmitterChain uint16 `json:"emitter_chain"` + } ) func (mo MissingObservation) String() string { return fmt.Sprintf("%d-%s", mo.ChainId, hex.EncodeToString(mo.TxHash)) } -// makeAuditKey creates an audit map key from a missing observation. -func (mo *MissingObservation) makeAuditKey() string { - return fmt.Sprintf("%d-%s", mo.ChainId, strings.TrimPrefix(hex.EncodeToString(mo.TxHash[:]), "0x")) -} - // makeAuditKey creates an audit map key from a pending observation entry. func (pe *pendingEntry) makeAuditKey() string { return fmt.Sprintf("%d-%s", pe.msg.EmitterChain, strings.TrimPrefix(pe.msg.TxIDString(), "0x")) } +// hasGuardianSigned checks if a guardian has signed based on the signatures bitmask. +// The signatures field is a u128 represented as a decimal string where each bit +// corresponds to a guardian index. +func hasGuardianSigned(signatures string, guardianIndex int) bool { + sigInt, ok := new(big.Int).SetString(signatures, 10) //nolint:mnd // Base 10 because the signatures are encoded as a decimal string + if !ok { + return false // Assume not signed if we can't parse + } + return sigInt.Bit(guardianIndex) == 1 +} + // audit is the runnable that executes the audit each interval. func (acct *Accountant) audit(ctx context.Context) error { ticker := time.NewTicker(auditInterval) @@ -121,21 +149,21 @@ func (acct *Accountant) audit(ctx context.Context) error { case <-ctx.Done(): return nil case <-ticker.C: - acct.runAudit() + acct.runAudit(ctx) } } } // runAudit is the entry point for the audit of the pending transfer map. It creates a temporary map of all pending transfers and invokes the main audit function. -func (acct *Accountant) runAudit() { - tmpMap := acct.createAuditMap(false) - acct.logger.Debug("in AuditPendingTransfers: starting base audit", zap.Int("numPending", numPendingEntries(tmpMap))) - acct.performAudit(tmpMap, acct.wormchainConn, acct.contract) +func (acct *Accountant) runAudit(ctx context.Context) { + knownPendingTransferMap := acct.createAuditMap(false) + acct.logger.Debug("in AuditPendingTransfers: starting base audit", zap.Int("numPending", len(knownPendingTransferMap))) + acct.performAudit(ctx, knownPendingTransferMap, acct.wormchainConn, acct.contract) acct.logger.Debug("in AuditPendingTransfers: finished base audit") - tmpMap = acct.createAuditMap(true) - acct.logger.Debug("in AuditPendingTransfers: starting ntt audit", zap.Int("numPending", numPendingEntries(tmpMap))) - acct.performAudit(tmpMap, acct.nttWormchainConn, acct.nttContract) + knownPendingTransferMap = acct.createAuditMap(true) + acct.logger.Debug("in AuditPendingTransfers: starting ntt audit", zap.Int("numPending", len(knownPendingTransferMap))) + acct.performAudit(ctx, knownPendingTransferMap, acct.nttWormchainConn, acct.nttContract) acct.logger.Debug("in AuditPendingTransfers: finished ntt audit") } @@ -160,7 +188,7 @@ func (acct *Accountant) createAuditMap(isNTT bool) map[string][]*pendingEntry { acct.pendingTransfersLock.Lock() defer acct.pendingTransfersLock.Unlock() - tmpMap := make(map[string][]*pendingEntry) + knownPendingTransferMap := make(map[string][]*pendingEntry) for _, pe := range acct.pendingTransfers { // Skip over nil entries if pe == nil { @@ -173,11 +201,11 @@ func (acct *Accountant) createAuditMap(isNTT bool) map[string][]*pendingEntry { } key := pe.makeAuditKey() acct.logger.Debug("will audit pending transfer", zap.String("msgId", pe.msgId), zap.String("moKey", key), zap.Bool("submitPending", pe.submitPending()), zap.Stringer("lastUpdateTime", pe.updTime())) - tmpMap[key] = append(tmpMap[key], pe) + knownPendingTransferMap[key] = append(knownPendingTransferMap[key], pe) } } - return tmpMap + return knownPendingTransferMap } // hasBeenPendingForTooLong determines if a transfer has been in the "submit pending" state for too long. @@ -189,12 +217,26 @@ func (pe *pendingEntry) hasBeenPendingForTooLong() bool { // performAudit audits the temporary map against the smart contract. It is meant to be run in a go routine. It takes a temporary map of all pending transfers // and validates that against what is reported by the smart contract. For more details, please see the prologue of this file. -func (acct *Accountant) performAudit(tmpMap map[string][]*pendingEntry, wormchainConn AccountantWormchainConn, contract string) { +func (acct *Accountant) performAudit(ctx context.Context, knownPendingTransferMap map[string][]*pendingEntry, wormchainConn AccountantWormchainConn, contract string) { acct.logger.Debug("entering performAudit", zap.String("contract", contract)) - missingObservations, err := acct.queryMissingObservations(wormchainConn, contract) + + gs := acct.gst.Get() + if gs == nil { + acct.logger.Error("unable to perform audit, failed to get guardian set") + return + } + + guardianIndex, found := gs.KeyIndex(acct.guardianAddr) + if !found { + acct.logger.Error("unable to perform audit, failed to get guardian index") + return + } + + // Query all pending transfers with pagination + pendingTransfers, err := acct.queryAllPendingTransfers(wormchainConn, contract) if err != nil { - acct.logger.Error("unable to perform audit, failed to query missing observations", zap.Error(err)) - for _, entries := range tmpMap { + acct.logger.Error("unable to perform audit, failed to query pending transfers", zap.Error(err)) + for _, entries := range knownPendingTransferMap { for _, pe := range entries { // We already do a nil check in `createAuditMap`, but we'll do the same here if pe == nil { @@ -206,102 +248,132 @@ func (acct *Accountant) performAudit(tmpMap map[string][]*pendingEntry, wormchai return } - for _, mo := range missingObservations { - key := mo.makeAuditKey() - entries, exists := tmpMap[key] - if exists { - for _, pe := range entries { - // We already do a nil check in `createAuditMap`, but we'll do the same here - if pe == nil { - continue - } - if acct.submitObservation(pe) { - auditErrors.Inc() - acct.logger.Error("contract reported pending observation as missing, resubmitted it", zap.String("msgID", pe.msgId)) - } else { - acct.logger.Info("contract reported pending observation as missing but it is queued up to be submitted, skipping it", zap.String("msgID", pe.msgId)) - } + acct.logger.Info("audit queried pending transfers", + zap.Int("totalPending", len(pendingTransfers)), + zap.String("contract", contract)) + + // SECURITY: knownPendingTransferMap contains only transfers that this guardian has personally + // observed and verified through the normal message processing pipeline. Transfers returned by + // the contract's all_pending_transfers query are untrusted external data. We must ONLY resubmit + // observations for transfers present in knownPendingTransferMap. For transfers the contract + // reports that we do NOT have locally, we request a reobservation from the watcher, which + // re-verifies the transaction on-chain before it enters the signing pipeline. Never construct + // a signable observation directly from contract response data. + for _, pt := range pendingTransfers { + for _, data := range pt.Data { + // Skip if we've already signed this + if hasGuardianSigned(data.Signatures, guardianIndex) { + continue } - delete(tmpMap, key) - } else { - acct.handleMissingObservation(mo) - } - } + // We haven't signed - build key to check our local map + key := fmt.Sprintf("%d-%s", data.EmitterChain, + strings.TrimPrefix(hex.EncodeToString(data.TxHash), "0x")) - if len(tmpMap) != 0 { - var keys []TransferKey - var pendingTransfers []*pendingEntry - for _, entries := range tmpMap { - for _, pe := range entries { - // We already do a nil check in `createAuditMap`, but we'll do the same here - if pe == nil { - continue - } - - keys = append(keys, TransferKey{EmitterChain: uint16(pe.msg.EmitterChain), EmitterAddress: pe.msg.EmitterAddress, Sequence: pe.msg.Sequence}) - pendingTransfers = append(pendingTransfers, pe) - } - } + if entries, exists := knownPendingTransferMap[key]; exists { + for _, pe := range entries { + // We already do a nil check in `createAuditMap`, but we'll do the same here + if pe == nil { + continue + } - transferDetails, err := acct.queryBatchTransferStatus(keys, wormchainConn, contract) - if err != nil { - acct.logger.Error("unable to finish audit, failed to query for transfer statuses", zap.Error(err)) - for _, pe := range pendingTransfers { - // We already do a nil check in `createAuditMap`, but we'll do the same here - if pe == nil { - continue + // We have it locally but haven't submitted successfully - resubmit + if acct.submitObservation(ctx, pe, true) { + auditErrors.Inc() + acct.logger.Error("contract reported we have not signed a pending transfer, resubmitting", zap.String("msgId", pe.msgId)) + } else { + acct.logger.Info("contract reported we have not signed a pending transfer but it is already pending submission, skipping", zap.String("msgId", pe.msgId)) + } + delete(knownPendingTransferMap, key) } - acct.logger.Error("unsure of status of pending transfer due to query error", zap.String("msgId", pe.msgId)) + } else { + // We don't have it locally - request reobservation + acct.handleMissingObservation(MissingObservation{ + ChainId: data.EmitterChain, + TxHash: data.TxHash, + }) } - return } + } - for _, pe := range pendingTransfers { - // There should be no nil entries, but we'll skip to be safe + if len(knownPendingTransferMap) == 0 { + acct.logger.Debug("exiting performAudit") + return + } + + // Anything still in knownPendingTransferMap is something WE have but the CONTRACT doesn't know about as pending. + // It could be committed (need to publish) or unknown (need to resubmit). + // Query the status to find out. + var keys []TransferKey + var localTransfers []*pendingEntry + for _, entries := range knownPendingTransferMap { + for _, pe := range entries { + // We already do a nil check in `createAuditMap`, but we'll do the same here if pe == nil { continue } - status, exists := transferDetails[pe.msgId] - if !exists { - if acct.submitObservation(pe) { - auditErrors.Inc() - acct.logger.Error("query did not return status for transfer, this should not happen, resubmitted it", zap.String("msgId", pe.msgId)) - } else { - acct.logger.Info("query did not return status for transfer we have not submitted yet, ignoring it", zap.String("msgId", pe.msgId)) - } + keys = append(keys, TransferKey{EmitterChain: uint16(pe.msg.EmitterChain), EmitterAddress: pe.msg.EmitterAddress, Sequence: pe.msg.Sequence}) + localTransfers = append(localTransfers, pe) + } + } + transferDetails, err := acct.queryBatchTransferStatus(keys, wormchainConn, contract) + if err != nil { + acct.logger.Error("unable to finish audit, failed to query for transfer statuses", zap.Error(err)) + for _, pe := range localTransfers { + // We already do a nil check in `createAuditMap`, but we'll do the same here + if pe == nil { continue } + acct.logger.Error("unsure of status of pending transfer due to query error", zap.String("msgId", pe.msgId)) + } + return + } - if status == nil { - // This is the case when the contract does not know about a transfer. Resubmit it. - if acct.submitObservation(pe) { - auditErrors.Inc() - acct.logger.Error("contract does not know about pending transfer, resubmitted it", zap.String("msgId", pe.msgId)) - } - } else if status.Committed != nil { - digest := hex.EncodeToString(status.Committed.Digest) - if pe.digest == digest { - acct.logger.Warn("audit determined that transfer has been committed, publishing it", zap.String("msgId", pe.msgId)) - acct.handleCommittedTransfer(pe.msgId) - } else { - digestMismatches.Inc() - acct.logger.Error("audit detected a digest mismatch, dropping transfer", zap.String("msgId", pe.msgId), zap.String("ourDigest", pe.digest), zap.String("reportedDigest", digest)) - acct.deletePendingTransfer(pe.msgId) - } - } else if status.Pending != nil { - acct.logger.Debug("contract says transfer is still pending", zap.String("msgId", pe.msgId)) + for _, pe := range localTransfers { + // There should be no nil entries, but we'll skip to be safe + if pe == nil { + continue + } + status, exists := transferDetails[pe.msgId] + if !exists { + if acct.submitObservation(ctx, pe, true) { + auditErrors.Inc() + acct.logger.Error("query did not return status for transfer, this should not happen, resubmitted it", zap.String("msgId", pe.msgId)) } else { - // This is the case when the contract does not know about a transfer. Resubmit it. - if acct.submitObservation(pe) { - auditErrors.Inc() - bytes, err := json.Marshal(*status) - if err != nil { - acct.logger.Error("unknown status returned for pending transfer, resubmitted it", zap.String("msgId", pe.msgId), zap.Error(err)) - } else { - acct.logger.Error("unknown status returned for pending transfer, resubmitted it", zap.String("msgId", pe.msgId), zap.String("status", string(bytes))) - } + acct.logger.Info("query did not return status for transfer we have not submitted yet, ignoring it", zap.String("msgId", pe.msgId)) + } + + continue + } + + if status == nil { + // This is the case when the contract does not know about a transfer. Resubmit it. + if acct.submitObservation(ctx, pe, true) { + auditErrors.Inc() + acct.logger.Error("contract does not know about pending transfer, resubmitted it", zap.String("msgId", pe.msgId)) + } + } else if status.Committed != nil { + digest := hex.EncodeToString(status.Committed.Digest) + if pe.digest == digest { + acct.logger.Warn("audit determined that transfer has been committed, publishing it", zap.String("msgId", pe.msgId)) + acct.handleCommittedTransfer(pe.msgId) + } else { + digestMismatches.Inc() + acct.logger.Error("audit detected a digest mismatch, dropping transfer", zap.String("msgId", pe.msgId), zap.String("ourDigest", pe.digest), zap.String("reportedDigest", digest)) + acct.deletePendingTransfer(pe.msgId) + } + } else if status.Pending != nil { + acct.logger.Debug("contract says transfer is still pending", zap.String("msgId", pe.msgId)) + } else { + // This is the case when the contract does not know about a transfer. Resubmit it. + if acct.submitObservation(ctx, pe, true) { + auditErrors.Inc() + bytes, err := json.Marshal(*status) + if err != nil { + acct.logger.Error("unknown status returned for pending transfer, resubmitted it", zap.String("msgId", pe.msgId), zap.Error(err)) + } else { + acct.logger.Error("unknown status returned for pending transfer, resubmitted it", zap.String("msgId", pe.msgId), zap.String("status", string(bytes))) } } } @@ -323,37 +395,85 @@ func (acct *Accountant) handleMissingObservation(mo MissingObservation) { } } -// queryMissingObservations queries the contract for the set of observations it thinks are missing for this guardian. -func (acct *Accountant) queryMissingObservations(wormchainConn AccountantWormchainConn, contract string) ([]MissingObservation, error) { - gs := acct.gst.Get() - if gs == nil { - return nil, fmt.Errorf("failed to get guardian set") +// queryConn allows us to mock the SubmitQuery call. +type queryConn interface { + SubmitQuery(ctx context.Context, contractAddress string, query []byte) ([]byte, error) +} + +// queryAllPendingTransfers paginates through all pending transfers from the contract. +func (acct *Accountant) queryAllPendingTransfers( + wormchainConn AccountantWormchainConn, + contract string, +) ([]PendingTransfer, error) { + return queryAllPendingTransfersWithConn(acct.ctx, acct.logger, wormchainConn, contract, allPendingTransfersPageSize) +} + +// queryAllPendingTransfersWithConn is a free function that paginates through all pending transfers. +// It accepts a queryConn interface to allow mocking in tests. +func queryAllPendingTransfersWithConn( + ctx context.Context, + logger *zap.Logger, + qc queryConn, + contract string, + pageSize int, +) ([]PendingTransfer, error) { + var allPending []PendingTransfer + var startAfter *TransferKey + + for { + page, err := queryAllPendingTransfersPage(ctx, logger, qc, contract, startAfter, pageSize) + if err != nil { + return nil, err + } + + allPending = append(allPending, page...) + + if len(page) < pageSize { + // Last page + break + } + + // Set cursor for next page + lastKey := page[len(page)-1].Key + startAfter = &lastKey } - guardianIndex, found := gs.KeyIndex(acct.guardianAddr) - if !found { - return nil, fmt.Errorf("failed to get guardian index") + return allPending, nil +} + +// queryAllPendingTransfersPage queries a single page of pending transfers. +func queryAllPendingTransfersPage( + ctx context.Context, + logger *zap.Logger, + qc queryConn, + contract string, + startAfter *TransferKey, + limit int, +) ([]PendingTransfer, error) { + var query string + if startAfter == nil { + query = fmt.Sprintf(`{"all_pending_transfers":{"limit":%d}}`, limit) + } else { + startAfterBytes, err := json.Marshal(startAfter) + if err != nil { + return nil, fmt.Errorf("failed to marshal start_after: %w", err) + } + query = fmt.Sprintf(`{"all_pending_transfers":{"start_after":%s,"limit":%d}}`, string(startAfterBytes), limit) } - query := fmt.Sprintf(`{"missing_observations":{"guardian_set": %d, "index": %d}}`, gs.Index, guardianIndex) - acct.logger.Debug("submitting missing_observations query", zap.String("query", query)) - respBytes, err := wormchainConn.SubmitQuery(acct.ctx, contract, []byte(query)) + logger.Debug("submitting all_pending_transfers query", zap.String("query", query)) + respBytes, err := qc.SubmitQuery(ctx, contract, []byte(query)) if err != nil { - return nil, fmt.Errorf("missing_observations query failed: %w, %s", err, query) + return nil, fmt.Errorf("all_pending_transfers query failed: %w, %s", err, query) } - var ret MissingObservationsResponse - if err := json.Unmarshal(respBytes, &ret); err != nil { - return nil, fmt.Errorf("failed to parse missing_observations response: %w, resp: %s", err, string(respBytes)) + var resp AllPendingTransfersResponse + if err := json.Unmarshal(respBytes, &resp); err != nil { + return nil, fmt.Errorf("failed to parse all_pending_transfers response: %w, resp: %s", err, string(respBytes)) } - acct.logger.Debug("missing_observations query response", zap.Int("numEntries", len(ret.Missing)), zap.String("result", string(respBytes))) - return ret.Missing, nil -} - -// queryConn allows us to mock the SubmitQuery call. -type queryConn interface { - SubmitQuery(ctx context.Context, contractAddress string, query []byte) ([]byte, error) + logger.Debug("all_pending_transfers query response", zap.Int("numEntries", len(resp.Pending))) + return resp.Pending, nil } // queryBatchTransferStatus queries the status of the specified transfers and returns a map keyed by transfer key (as a string) to the status. diff --git a/node/pkg/accountant/data_for_test.go b/node/pkg/accountant/data_for_test.go index 7cf2e7c3c78..7cbe89a302e 100644 --- a/node/pkg/accountant/data_for_test.go +++ b/node/pkg/accountant/data_for_test.go @@ -1,6 +1,7 @@ package accountant import ( + "encoding/base64" "fmt" "testing" @@ -5414,3 +5415,125 @@ func createTxRespForCommitted() []byte { return []byte(respJson) } + +// createPendingTransfersForTest creates test JSON data for all_pending_transfers queries. +// It reuses the same transfer key data as createTransferKeysForTestingBatchTransferStatus. +// Parameters: +// - num: number of transfers to create (-1 for all available) +// - guardianSetIndex: the guardian set index to use for all transfers +// - signatures: the signatures bitmask string (decimal representation of u128) +// +// Returns the JSON response bytes. +func createPendingTransfersForTest(t *testing.T, num int, guardianSetIndex uint32, signatures string) []byte { //nolint:unparam + input := []transferKey{ + { + EmitterChain: 1, + EmitterAddress: "ec7372995d5cc8732397fb0ad35c0121e0eaa90d26f828a534cab54391b3a4f5", + Sequence: 277025, + }, + { + EmitterChain: 3, + EmitterAddress: "0000000000000000000000007cf7b764e38a0a5e967972c1df77d432510564e2", + Sequence: 258114, + }, + { + EmitterChain: 2, + EmitterAddress: "0000000000000000000000003ee18b2214aff97000d974cf647e7c347e8fa585", + Sequence: 106099, + }, + { + EmitterChain: 1, + EmitterAddress: "ec7372995d5cc8732397fb0ad35c0121e0eaa90d26f828a534cab54391b3a4f5", + Sequence: 277276, + }, + { + EmitterChain: 2, + EmitterAddress: "0000000000000000000000003ee18b2214aff97000d974cf647e7c347e8fa585", + Sequence: 106014, + }, + } + + if num >= 0 { + require.GreaterOrEqual(t, len(input), num) + input = input[:num] + } + + respBytes := []byte("{\"pending\":[") + first := true + for i, in := range input { + emitterAddr, _ := vaa.StringToAddress(in.EmitterAddress) + + // Create a simple digest and tx_hash based on the index for deterministic testing + digest := fmt.Sprintf("digest%d", i) + txHash := fmt.Sprintf("txhash%d", i) + + // Build JSON - note that digest and tx_hash are base64 encoded in JSON + digestB64 := base64.StdEncoding.EncodeToString([]byte(digest)) + txHashB64 := base64.StdEncoding.EncodeToString([]byte(txHash)) + bytes := fmt.Sprintf( + `{"key":{"emitter_chain":%d,"emitter_address":"%s","sequence":%d},"data":[{"digest":"%s","tx_hash":"%s","signatures":"%s","guardian_set_index":%d,"emitter_chain":%d}]}`, + in.EmitterChain, emitterAddr.String(), in.Sequence, + digestB64, txHashB64, signatures, guardianSetIndex, in.EmitterChain) + + if first { + first = false + } else { + respBytes = append(respBytes, ',') + } + respBytes = append(respBytes, bytes...) + } + + respBytes = append(respBytes, []byte("]}")...) + return respBytes +} + +// createPendingTransfersForTestWithTxHash creates test JSON data for all_pending_transfers +// with a specific txHash that can be matched against local pending entries. +func createPendingTransfersForTestWithTxHash( + emitterChain uint16, //nolint:unparam + emitterAddr string, + sequence uint64, + txHash []byte, + digest []byte, + guardianSetIndex uint32, + signatures string, +) []byte { + emitterAddress, _ := vaa.StringToAddress(emitterAddr) + txHashB64 := base64.StdEncoding.EncodeToString(txHash) + digestB64 := base64.StdEncoding.EncodeToString(digest) + + return []byte(fmt.Sprintf( + `{"pending":[{"key":{"emitter_chain":%d,"emitter_address":"%s","sequence":%d},"data":[{"digest":"%s","tx_hash":"%s","signatures":"%s","guardian_set_index":%d,"emitter_chain":%d}]}]}`, + emitterChain, emitterAddress.String(), sequence, + digestB64, txHashB64, signatures, guardianSetIndex, emitterChain)) +} + +// createBatchTransferStatusResponse creates a batch_transfer_status JSON response. +// statusType: "committed", "pending", or "null" +// For "committed", digest is required (raw bytes, will be base64 encoded) +func createBatchTransferStatusResponse( + emitterChain uint16, + emitterAddr string, + sequence uint64, + statusType string, + digest []byte, +) []byte { + emitterAddress, _ := vaa.StringToAddress(emitterAddr) + + var statusJson string + switch statusType { + case "committed": + digestB64 := base64.StdEncoding.EncodeToString(digest) + statusJson = fmt.Sprintf(`{"committed":{"data":{"amount":"1000000000000000000","token_chain":2,"token_address":"0000000000000000000000002d8be6bf0baa74e0a907016679cae9190e80dd0a","recipient_chain":4},"digest":"%s"}}`, digestB64) + case "pending": + statusJson = `{"pending":[{"digest":"dGVzdA==","tx_hash":"dGVzdA==","signatures":"1","guardian_set_index":0,"emitter_chain":2}]}` + case "null": + statusJson = "null" + default: + statusJson = "null" + } + + return []byte(fmt.Sprintf( + `{"details":[{"key":{"emitter_chain":%d,"emitter_address":"%s","sequence":%d},"status":%s}]}`, + emitterChain, emitterAddress.String(), sequence, statusJson)) +} diff --git a/node/pkg/accountant/metrics.go b/node/pkg/accountant/metrics.go index 12eb073ac40..ac17496b471 100644 --- a/node/pkg/accountant/metrics.go +++ b/node/pkg/accountant/metrics.go @@ -57,4 +57,9 @@ var ( Name: "global_accountant_audit_errors_total", Help: "Total number of audit errors detected by accountant", }) + channelSubmitTimeouts = promauto.NewCounter( + prometheus.CounterOpts{ + Name: "global_accountant_channel_submit_timeouts", + Help: "Total number of channel submit timeouts during audit", + }) ) diff --git a/node/pkg/accountant/query_test.go b/node/pkg/accountant/query_test.go index e7fc750ffb6..5cc593f3160 100644 --- a/node/pkg/accountant/query_test.go +++ b/node/pkg/accountant/query_test.go @@ -18,35 +18,6 @@ import ( "go.uber.org/zap" ) -func TestParseMissingObservationsResponse(t *testing.T) { - responsesJson := []byte("{\"missing\":[{\"chain_id\":2,\"tx_hash\":\"y1E+jwgozWKEVbHt9dIeErgNXlHnntQwdzymYSCmBEA=\"},{\"chain_id\":4,\"tx_hash\":\"FZyF7xR5bIwtvdBIlIrDEZc+mrCkN/ixjGazgJdJdQQ=\"}]}") - var response MissingObservationsResponse - err := json.Unmarshal(responsesJson, &response) - require.NoError(t, err) - require.Equal(t, 2, len(response.Missing)) - - expectedTxHash0, err := hex.DecodeString("cb513e8f0828cd628455b1edf5d21e12b80d5e51e79ed430773ca66120a60440") - require.NoError(t, err) - - expectedTxHash1, err := hex.DecodeString("159c85ef14796c8c2dbdd048948ac311973e9ab0a437f8b18c66b38097497504") - require.NoError(t, err) - - expectedResult := MissingObservationsResponse{ - Missing: []MissingObservation{ - MissingObservation{ - ChainId: uint16(vaa.ChainIDEthereum), - TxHash: expectedTxHash0, - }, - MissingObservation{ - ChainId: uint16(vaa.ChainIDBSC), - TxHash: expectedTxHash1, - }, - }, - } - - assert.Equal(t, expectedResult, response) -} - func TestParseBatchTransferStatusCommittedResponse(t *testing.T) { responsesJson := []byte("{\"details\":[{\"key\":{\"emitter_chain\":2,\"emitter_address\":\"0000000000000000000000000290fb167208af455bb137780163b7b7a9a10c16\",\"sequence\":1674568234},\"status\":{\"committed\":{\"data\":{\"amount\":\"1000000000000000000\",\"token_chain\":2,\"token_address\":\"0000000000000000000000002d8be6bf0baa74e0a907016679cae9190e80dd0a\",\"recipient_chain\":4},\"digest\":\"1nbbff/7/ai9GJUs4h2JymFuO4+XcasC6t05glXc99M=\"}}}]}") var response BatchTransferStatusResponse @@ -232,3 +203,234 @@ func TestBatchTransferStatusMultipleChunks(t *testing.T) { require.Equal(t, len(keys), len(transferDetails)) validateBatchTransferStatusResults(t, keys, transferDetails) } + +func TestHasGuardianSigned(t *testing.T) { + tests := []struct { + name string + signatures string + guardianIndex int + expectedResult bool + }{ + { + name: "guardian 0 signed (bit 0 set)", + signatures: "1", + guardianIndex: 0, + expectedResult: true, + }, + { + name: "guardian 0 not signed (bit 0 not set)", + signatures: "2", + guardianIndex: 0, + expectedResult: false, + }, + { + name: "guardian 1 signed (bit 1 set)", + signatures: "2", + guardianIndex: 1, + expectedResult: true, + }, + { + name: "guardian 0 and 1 signed (bits 0 and 1 set)", + signatures: "3", + guardianIndex: 0, + expectedResult: true, + }, + { + name: "guardian 0 and 1 signed, check guardian 1", + signatures: "3", + guardianIndex: 1, + expectedResult: true, + }, + { + name: "guardian 0 and 1 signed, check guardian 2", + signatures: "3", + guardianIndex: 2, + expectedResult: false, + }, + { + name: "all 19 guardians signed", + signatures: "524287", // 2^19 - 1 = 0x7FFFF + guardianIndex: 18, + expectedResult: true, + }, + { + name: "all 19 guardians signed, check guardian 19", + signatures: "524287", + guardianIndex: 19, + expectedResult: false, + }, + { + name: "large signature value with guardian 63 signed", + signatures: "9223372036854775808", // 2^63 + guardianIndex: 63, + expectedResult: true, + }, + { + name: "empty signatures string", + signatures: "", + guardianIndex: 0, + expectedResult: false, + }, + { + name: "invalid signatures string", + signatures: "not-a-number", + guardianIndex: 0, + expectedResult: false, + }, + { + name: "zero signatures", + signatures: "0", + guardianIndex: 0, + expectedResult: false, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + result := hasGuardianSigned(tc.signatures, tc.guardianIndex) + assert.Equal(t, tc.expectedResult, result) + }) + } +} + +func TestParseAllPendingTransfersResponse(t *testing.T) { + responsesJson := []byte(`{"pending":[{"key":{"emitter_chain":2,"emitter_address":"0000000000000000000000000290fb167208af455bb137780163b7b7a9a10c16","sequence":1674568234},"data":[{"digest":"1nbbff/7/ai9GJUs4h2JymFuO4+XcasC6t05glXc99M=","tx_hash":"CjHx8zExnr4JU8ewAu5/tXM6a5QyslKufGHZNSr0aE8=","signatures":"3","guardian_set_index":0,"emitter_chain":2}]}]}`) + var response AllPendingTransfersResponse + err := json.Unmarshal(responsesJson, &response) + require.NoError(t, err) + require.Equal(t, 1, len(response.Pending)) + + expectedEmitterAddress, err := vaa.StringToAddress("0000000000000000000000000290fb167208af455bb137780163b7b7a9a10c16") + require.NoError(t, err) + + expectedDigest, err := hex.DecodeString("d676db7dfffbfda8bd18952ce21d89ca616e3b8f9771ab02eadd398255dcf7d3") + require.NoError(t, err) + + expectedTxHash, err := hex.DecodeString("0a31f1f331319ebe0953c7b002ee7fb5733a6b9432b252ae7c61d9352af4684f") + require.NoError(t, err) + + expectedResult := PendingTransfer{ + Key: TransferKey{ + EmitterChain: uint16(vaa.ChainIDEthereum), + EmitterAddress: expectedEmitterAddress, + Sequence: 1674568234, + }, + Data: []PendingTransferData{ + { + Digest: expectedDigest, + TxHash: expectedTxHash, + Signatures: "3", + GuardianSetIndex: 0, + EmitterChain: uint16(vaa.ChainIDEthereum), + }, + }, + } + + assert.True(t, reflect.DeepEqual(expectedResult, response.Pending[0])) +} + +// AllPendingTransfersQueryConnMock allows us to mock all_pending_transfers by implementing SubmitQuery. +type AllPendingTransfersQueryConnMock struct { + // pages holds the response data for each page, keyed by page index + pages [][]byte + pageIndex int + err error +} + +func (qc *AllPendingTransfersQueryConnMock) SubmitQuery(ctx context.Context, contractAddress string, query []byte) ([]byte, error) { + if qc.err != nil { + return nil, qc.err + } + + if qc.pageIndex >= len(qc.pages) { + // Return empty response if we've exhausted pages + return []byte(`{"pending":[]}`), nil + } + + resp := qc.pages[qc.pageIndex] + qc.pageIndex++ + return resp, nil +} + +func TestQueryAllPendingTransfersPage(t *testing.T) { + ctx := context.Background() + logger := zap.NewNop() + + queryResp := createPendingTransfersForTest(t, 3, 0, "1") + qc := &AllPendingTransfersQueryConnMock{pages: [][]byte{queryResp}} + + pending, err := queryAllPendingTransfersPage(ctx, logger, qc, "wormhole14hj2tavq8fpesdwxxcu44rty3hh90vhujrvcmstl4zr3txmfvw9srrg465", nil, 500) + require.NoError(t, err) + require.Equal(t, 3, len(pending)) +} + +func TestQueryAllPendingTransfersEmpty(t *testing.T) { + ctx := context.Background() + logger := zap.NewNop() + + qc := &AllPendingTransfersQueryConnMock{pages: [][]byte{[]byte(`{"pending":[]}`)}} + + pending, err := queryAllPendingTransfersWithConn(ctx, logger, qc, "wormhole14hj2tavq8fpesdwxxcu44rty3hh90vhujrvcmstl4zr3txmfvw9srrg465", 500) + require.NoError(t, err) + require.Equal(t, 0, len(pending)) +} + +func TestQueryAllPendingTransfersPagination(t *testing.T) { + ctx := context.Background() + logger := zap.NewNop() + + page1 := createPendingTransfersForTest(t, 3, 0, "1") + page2 := createPendingTransfersForTest(t, 2, 0, "1") + + qc := &AllPendingTransfersQueryConnMock{pages: [][]byte{page1, page2}} + + // Use pageSize=3 so first page is "full" and triggers fetching second page + pending, err := queryAllPendingTransfersWithConn(ctx, logger, qc, "wormhole14hj2tavq8fpesdwxxcu44rty3hh90vhujrvcmstl4zr3txmfvw9srrg465", 3) + require.NoError(t, err) + require.Equal(t, 5, len(pending)) // 3 from page1 + 2 from page2 +} + +func TestQueryAllPendingTransfersPaginationMultipleFullPages(t *testing.T) { + ctx := context.Background() + logger := zap.NewNop() + + page1 := createPendingTransfersForTest(t, 3, 0, "1") + page2 := createPendingTransfersForTest(t, 3, 0, "1") + emptyPage := []byte(`{"pending":[]}`) + + qc := &AllPendingTransfersQueryConnMock{pages: [][]byte{page1, page2, emptyPage}} + + // Use pageSize=3 so first two pages are "full" and trigger fetching next page + // Third page is empty which terminates pagination + pending, err := queryAllPendingTransfersWithConn(ctx, logger, qc, "wormhole14hj2tavq8fpesdwxxcu44rty3hh90vhujrvcmstl4zr3txmfvw9srrg465", 3) + require.NoError(t, err) + require.Equal(t, 6, len(pending)) // 3 from page1 + 3 from page2 + 0 from empty page +} + +func TestQueryAllPendingTransfersQueryError(t *testing.T) { + ctx := context.Background() + logger := zap.NewNop() + + qc := &AllPendingTransfersQueryConnMock{err: errors.New("query failed")} + + _, err := queryAllPendingTransfersWithConn(ctx, logger, qc, "wormhole14hj2tavq8fpesdwxxcu44rty3hh90vhujrvcmstl4zr3txmfvw9srrg465", 500) + require.Error(t, err) + assert.Contains(t, err.Error(), "query failed") +} + +func TestQueryAllPendingTransfersMultipleDataEntries(t *testing.T) { + // Test parsing a response where a single pending transfer has multiple data entries + // (which can happen when the same transfer has observations from different guardian sets) + responsesJson := []byte(`{"pending":[{"key":{"emitter_chain":2,"emitter_address":"0000000000000000000000000290fb167208af455bb137780163b7b7a9a10c16","sequence":100},"data":[{"digest":"AAAA","tx_hash":"BBBB","signatures":"1","guardian_set_index":0,"emitter_chain":2},{"digest":"CCCC","tx_hash":"DDDD","signatures":"2","guardian_set_index":1,"emitter_chain":2}]}]}`) + + var response AllPendingTransfersResponse + err := json.Unmarshal(responsesJson, &response) + require.NoError(t, err) + require.Equal(t, 1, len(response.Pending)) + require.Equal(t, 2, len(response.Pending[0].Data)) + + assert.Equal(t, uint32(0), response.Pending[0].Data[0].GuardianSetIndex) + assert.Equal(t, "1", response.Pending[0].Data[0].Signatures) + assert.Equal(t, uint32(1), response.Pending[0].Data[1].GuardianSetIndex) + assert.Equal(t, "2", response.Pending[0].Data[1].Signatures) +} diff --git a/node/pkg/accountant/submit_obs.go b/node/pkg/accountant/submit_obs.go index f7020505c5a..b688b839302 100644 --- a/node/pkg/accountant/submit_obs.go +++ b/node/pkg/accountant/submit_obs.go @@ -22,8 +22,8 @@ import ( ) const ( - batchSize = 10 - batchTimeout = 100 * time.Millisecond + batchSize = 100 // Observations per batch (limited by wasm contract input size of 64KB) + batchTimeout = 2 * time.Second // Time to collect observations before submitting ) // baseWorker is the entry point for the base accountant worker. diff --git a/node/pkg/accountant/watcher.go b/node/pkg/accountant/watcher.go index 6f336f7cf91..a58c0ac11b3 100644 --- a/node/pkg/accountant/watcher.go +++ b/node/pkg/accountant/watcher.go @@ -62,7 +62,7 @@ func (acct *Accountant) watcher(ctx context.Context, isNTT bool) error { ctx, "guardiand", query, - 64, // channel capacity + 1000, // channel capacity ) if err != nil { return fmt.Errorf("failed to subscribe to %s events: %w", tag, err) @@ -211,7 +211,8 @@ func (acct *Accountant) processPendingTransfer(xfer *WasmObservation, tag string acct.publishTransferAlreadyLocked(pe) transfersApproved.Inc() } else { - // TODO: We could issue a reobservation request here since it looks like other guardians have seen this transfer but we haven't. + // This log will be emitted by the Guardians that submit to the Accountant after the transfer has already been confirmed. + // These transfers are already processed in submit_obs.go and so when the watcher sees the event it is no longer in the pendingTransfers map. acct.logger.Info("acctwatch: unknown transfer has been approved, ignoring it", zap.String("msgId", msgId)) } } diff --git a/node/pkg/wormconn/send_tx.go b/node/pkg/wormconn/send_tx.go index b65cf025265..f7e544296a3 100644 --- a/node/pkg/wormconn/send_tx.go +++ b/node/pkg/wormconn/send_tx.go @@ -35,7 +35,7 @@ func (c *ClientConn) SignAndBroadcastTx(ctx context.Context, msg sdktypes.Msg) ( if setErr := builder.SetMsgs(msg); setErr != nil { return nil, fmt.Errorf("failed to add message to builder: %w", setErr) } - builder.SetGasLimit(2000000) // TODO: Maybe simulate and use the result + builder.SetGasLimit(5000000) // Can comfortably handle 100 observations at ~4.2M gas maximum // The tx needs to be signed in 2 passes: first we populate the SignerInfo // inside the TxBuilder and then sign the payload. From db2230827b62a872348407872825b633ae2a132e Mon Sep 17 00:00:00 2001 From: Dirk Brink Date: Tue, 26 May 2026 15:28:56 -0600 Subject: [PATCH 2/6] Address Claude comments --- node/pkg/accountant/accountant.go | 3 ++- node/pkg/accountant/audit.go | 7 +++---- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/node/pkg/accountant/accountant.go b/node/pkg/accountant/accountant.go index 48aa09e4aa8..e95283881d3 100644 --- a/node/pkg/accountant/accountant.go +++ b/node/pkg/accountant/accountant.go @@ -449,14 +449,15 @@ func (acct *Accountant) loadPendingTransfers() error { // block until the channel has space, a timeout occurs, or the context is cancelled. This function grabs the state lock. func (acct *Accountant) submitObservation(ctx context.Context, pe *pendingEntry, blocking bool) bool { pe.stateLock.Lock() - defer pe.stateLock.Unlock() if pe.state.submitPending { + pe.stateLock.Unlock() return false } pe.state.submitPending = true pe.state.updTime = time.Now() + pe.stateLock.Unlock() timeout := time.Duration(0) if blocking { diff --git a/node/pkg/accountant/audit.go b/node/pkg/accountant/audit.go index 5d37939017e..ea0c2df4ef2 100644 --- a/node/pkg/accountant/audit.go +++ b/node/pkg/accountant/audit.go @@ -2,7 +2,6 @@ // It uses a ticker to periodically run the audit. The audit occurs in two phases that operate off of a temporary map of all pending transfers known to this guardian. // // The first phase involves querying the smart contract for all pending transfers using the "all_pending_transfers" query. For each pending transfer returned: -// - If the transfer is for a different guardian set, it is skipped. // - If this guardian has already signed (based on the signatures bitmask), it is skipped. // - If the transfer is in our local pending map, we resubmit our observation to the contract. // - If the transfer is not in our local map, we request a reobservation from the local watcher. @@ -157,12 +156,12 @@ func (acct *Accountant) audit(ctx context.Context) error { // runAudit is the entry point for the audit of the pending transfer map. It creates a temporary map of all pending transfers and invokes the main audit function. func (acct *Accountant) runAudit(ctx context.Context) { knownPendingTransferMap := acct.createAuditMap(false) - acct.logger.Debug("in AuditPendingTransfers: starting base audit", zap.Int("numPending", len(knownPendingTransferMap))) + acct.logger.Debug("in AuditPendingTransfers: starting base audit", zap.Int("numPending", numPendingEntries(knownPendingTransferMap))) acct.performAudit(ctx, knownPendingTransferMap, acct.wormchainConn, acct.contract) acct.logger.Debug("in AuditPendingTransfers: finished base audit") knownPendingTransferMap = acct.createAuditMap(true) - acct.logger.Debug("in AuditPendingTransfers: starting ntt audit", zap.Int("numPending", len(knownPendingTransferMap))) + acct.logger.Debug("in AuditPendingTransfers: starting ntt audit", zap.Int("numPending", numPendingEntries(knownPendingTransferMap))) acct.performAudit(ctx, knownPendingTransferMap, acct.nttWormchainConn, acct.nttContract) acct.logger.Debug("in AuditPendingTransfers: finished ntt audit") } @@ -284,8 +283,8 @@ func (acct *Accountant) performAudit(ctx context.Context, knownPendingTransferMa } else { acct.logger.Info("contract reported we have not signed a pending transfer but it is already pending submission, skipping", zap.String("msgId", pe.msgId)) } - delete(knownPendingTransferMap, key) } + delete(knownPendingTransferMap, key) } else { // We don't have it locally - request reobservation acct.handleMissingObservation(MissingObservation{ From 6573a5a28985007b027ff03502f6b22ec4623b17 Mon Sep 17 00:00:00 2001 From: Dirk Brink Date: Wed, 27 May 2026 10:45:21 -0600 Subject: [PATCH 3/6] Update comment and acquire lock in called function --- node/pkg/accountant/accountant.go | 6 ++++++ node/pkg/accountant/audit.go | 3 +-- 2 files changed, 7 insertions(+), 2 deletions(-) diff --git a/node/pkg/accountant/accountant.go b/node/pkg/accountant/accountant.go index e95283881d3..f8f89475a84 100644 --- a/node/pkg/accountant/accountant.go +++ b/node/pkg/accountant/accountant.go @@ -486,10 +486,14 @@ func (acct *Accountant) submitToChannel(ctx context.Context, pe *pendingEntry, s acct.logger.Warn(fmt.Sprintf("timeout submitting observation to %s channel, will retry next audit", tag), zap.String("msgId", pe.msgId), zap.Duration("timeout", timeout)) + pe.stateLock.Lock() pe.state.submitPending = false + pe.stateLock.Unlock() case <-ctx.Done(): acct.logger.Debug(fmt.Sprintf("context cancelled while submitting to %s channel", tag), zap.String("msgId", pe.msgId)) + pe.stateLock.Lock() pe.state.submitPending = false + pe.stateLock.Unlock() } } else { // Non-blocking write (existing behavior for processor) @@ -498,7 +502,9 @@ func (acct *Accountant) submitToChannel(ctx context.Context, pe *pendingEntry, s acct.logger.Debug(fmt.Sprintf("submitted observation to channel for %s", tag), zap.String("msgId", pe.msgId)) default: acct.logger.Error(fmt.Sprintf("unable to submit observation to %s because the channel is full, will try next interval", tag), zap.String("msgId", pe.msgId)) + pe.stateLock.Lock() pe.state.submitPending = false + pe.stateLock.Unlock() } } } diff --git a/node/pkg/accountant/audit.go b/node/pkg/accountant/audit.go index ea0c2df4ef2..42c710b680d 100644 --- a/node/pkg/accountant/audit.go +++ b/node/pkg/accountant/audit.go @@ -251,8 +251,7 @@ func (acct *Accountant) performAudit(ctx context.Context, knownPendingTransferMa zap.Int("totalPending", len(pendingTransfers)), zap.String("contract", contract)) - // SECURITY: knownPendingTransferMap contains only transfers that this guardian has personally - // observed and verified through the normal message processing pipeline. Transfers returned by + // SECURITY: knownPendingTransferMap contains only transfers that are verified through the normal message processing pipeline. Transfers returned by // the contract's all_pending_transfers query are untrusted external data. We must ONLY resubmit // observations for transfers present in knownPendingTransferMap. For transfers the contract // reports that we do NOT have locally, we request a reobservation from the watcher, which From 8990c42581d945a4f05026300ed6ce9af1cfb603 Mon Sep 17 00:00:00 2001 From: Dirk Brink Date: Tue, 30 Jun 2026 15:17:09 -0600 Subject: [PATCH 4/6] More minor PR feedback --- node/pkg/accountant/accountant.go | 4 ++-- node/pkg/accountant/audit.go | 8 ++++---- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/node/pkg/accountant/accountant.go b/node/pkg/accountant/accountant.go index f8f89475a84..65207bddb6a 100644 --- a/node/pkg/accountant/accountant.go +++ b/node/pkg/accountant/accountant.go @@ -490,13 +490,13 @@ func (acct *Accountant) submitToChannel(ctx context.Context, pe *pendingEntry, s pe.state.submitPending = false pe.stateLock.Unlock() case <-ctx.Done(): - acct.logger.Debug(fmt.Sprintf("context cancelled while submitting to %s channel", tag), zap.String("msgId", pe.msgId)) + acct.logger.Warn(fmt.Sprintf("context cancelled while submitting to %s channel", tag), zap.String("msgId", pe.msgId)) pe.stateLock.Lock() pe.state.submitPending = false pe.stateLock.Unlock() } } else { - // Non-blocking write (existing behavior for processor) + // Non-blocking write select { case subChan <- pe.msg: acct.logger.Debug(fmt.Sprintf("submitted observation to channel for %s", tag), zap.String("msgId", pe.msgId)) diff --git a/node/pkg/accountant/audit.go b/node/pkg/accountant/audit.go index 42c710b680d..32b87cf173e 100644 --- a/node/pkg/accountant/audit.go +++ b/node/pkg/accountant/audit.go @@ -160,9 +160,9 @@ func (acct *Accountant) runAudit(ctx context.Context) { acct.performAudit(ctx, knownPendingTransferMap, acct.wormchainConn, acct.contract) acct.logger.Debug("in AuditPendingTransfers: finished base audit") - knownPendingTransferMap = acct.createAuditMap(true) - acct.logger.Debug("in AuditPendingTransfers: starting ntt audit", zap.Int("numPending", numPendingEntries(knownPendingTransferMap))) - acct.performAudit(ctx, knownPendingTransferMap, acct.nttWormchainConn, acct.nttContract) + knownPendingNttTransferMap := acct.createAuditMap(true) + acct.logger.Debug("in AuditPendingTransfers: starting ntt audit", zap.Int("numPending", numPendingEntries(knownPendingNttTransferMap))) + acct.performAudit(ctx, knownPendingNttTransferMap, acct.nttWormchainConn, acct.nttContract) acct.logger.Debug("in AuditPendingTransfers: finished ntt audit") } @@ -295,7 +295,7 @@ func (acct *Accountant) performAudit(ctx context.Context, knownPendingTransferMa } if len(knownPendingTransferMap) == 0 { - acct.logger.Debug("exiting performAudit") + acct.logger.Debug("exiting performAudit with no known pending transfers left") return } From a7fb15f8beb0504b0d0cbea639a58b348444cb7d Mon Sep 17 00:00:00 2001 From: Dirk Brink Date: Tue, 30 Jun 2026 15:41:47 -0600 Subject: [PATCH 5/6] Add guardian arg to change batch size --- node/cmd/guardiand/node.go | 15 +++-- node/pkg/accountant/accountant.go | 77 ++++++++++++++------------ node/pkg/accountant/accountant_test.go | 2 + node/pkg/accountant/submit_obs.go | 6 +- node/pkg/node/node_test.go | 17 ++++++ node/pkg/node/options.go | 8 ++- 6 files changed, 80 insertions(+), 45 deletions(-) diff --git a/node/cmd/guardiand/node.go b/node/cmd/guardiand/node.go index 6ded740aef1..414c0aae0bd 100644 --- a/node/cmd/guardiand/node.go +++ b/node/cmd/guardiand/node.go @@ -15,6 +15,7 @@ import ( "time" "github.com/btcsuite/btcd/btcec/v2" + "github.com/certusone/wormhole/node/pkg/accountant" "github.com/certusone/wormhole/node/pkg/guardiansigner" "github.com/certusone/wormhole/node/pkg/watchers" "github.com/certusone/wormhole/node/pkg/watchers/ibc" @@ -129,11 +130,12 @@ var ( ibcBlockHeightURL *string ibcContract *string - accountantContract *string - accountantWS *string - accountantCheckEnabled *bool - accountantKeyPath *string - accountantKeyPassPhrase *string + accountantContract *string + accountantWS *string + accountantCheckEnabled *bool + accountantKeyPath *string + accountantKeyPassPhrase *string + accountantSubmitObservationBatchSize *int accountantNttContract *string accountantNttKeyPath *string @@ -404,6 +406,7 @@ func init() { accountantKeyPath = NodeCmd.Flags().String("accountantKeyPath", "", "path to accountant private key for signing transactions") accountantKeyPassPhrase = NodeCmd.Flags().String("accountantKeyPassPhrase", "", "pass phrase used to unarmor the accountant key file") accountantCheckEnabled = NodeCmd.Flags().Bool("accountantCheckEnabled", false, "Should accountant be enforced on transfers") + accountantSubmitObservationBatchSize = NodeCmd.Flags().Int("accountantSubmitObservationBatchSize", accountant.DefaultSubmitObservationBatchSize, "maximum number of observations to submit to the accountant contract in one transaction") accountantNttContract = NodeCmd.Flags().String("accountantNttContract", "", "Address of the NTT accountant smart contract on wormchain") accountantNttKeyPath = NodeCmd.Flags().String("accountantNttKeyPath", "", "path to NTT accountant private key for signing transactions") @@ -2055,7 +2058,7 @@ func runNode(cmd *cobra.Command, args []string) { guardianOptions := []*node.GuardianOption{ node.GuardianOptionDatabase(db), node.GuardianOptionWatchers(watcherConfigs, ibcWatcherConfig), - node.GuardianOptionAccountant(*accountantWS, *accountantContract, *accountantCheckEnabled, accountantWormchainConn, *accountantNttContract, accountantNttWormchainConn), + node.GuardianOptionAccountant(*accountantWS, *accountantContract, *accountantCheckEnabled, accountantWormchainConn, *accountantNttContract, accountantNttWormchainConn, *accountantSubmitObservationBatchSize), node.GuardianOptionGovernor(*chainGovernorEnabled, *governorFlowCancelEnabled, *coinGeckoApiKey), node.GuardianOptionNotary(*notaryEnabled), node.GuardianOptionManagerService(*managerServiceEnabled, managerSigners, *ethRPC), diff --git a/node/pkg/accountant/accountant.go b/node/pkg/accountant/accountant.go index 65207bddb6a..013af7e9c11 100644 --- a/node/pkg/accountant/accountant.go +++ b/node/pkg/accountant/accountant.go @@ -30,7 +30,7 @@ import ( // MsgChannelCapacity specifies the capacity of the message channel used to publish messages released from the accountant. // This channel should not back up, but if it does, the accountant will start dropping messages, which would require reobservations. -const MsgChannelCapacity = 5 * batchSize +const MsgChannelCapacity = 5 * DefaultSubmitObservationBatchSize type ( AccountantWormchainConn interface { @@ -75,23 +75,24 @@ type ( // Accountant is the object that manages the interface to the wormchain accountant smart contract. type Accountant struct { - ctx context.Context - logger *zap.Logger - db guardianDB.AccountantDB - obsvReqWriteC chan<- *gossipv1.ObservationRequest - contract string - wsUrl string - wormchainConn AccountantWormchainConn - enforceFlag bool - guardianSigner guardiansigner.GuardianSigner - gst *common.GuardianSetState - guardianAddr ethCommon.Address - msgChan chan<- *common.MessagePublication - tokenBridges validEmitters - pendingTransfersLock sync.Mutex - pendingTransfers map[string]*pendingEntry // Key is the message ID (emitterChain/emitterAddr/seqNo) - subChan chan *common.MessagePublication - env common.Environment + ctx context.Context + logger *zap.Logger + db guardianDB.AccountantDB + obsvReqWriteC chan<- *gossipv1.ObservationRequest + contract string + wsUrl string + wormchainConn AccountantWormchainConn + enforceFlag bool + guardianSigner guardiansigner.GuardianSigner + gst *common.GuardianSetState + guardianAddr ethCommon.Address + msgChan chan<- *common.MessagePublication + tokenBridges validEmitters + pendingTransfersLock sync.Mutex + pendingTransfers map[string]*pendingEntry // Key is the message ID (emitterChain/emitterAddr/seqNo) + subChan chan *common.MessagePublication + submitObservationBatchSize int + env common.Environment nttContract string nttWormchainConn AccountantWormchainConn @@ -126,25 +127,31 @@ func NewAccountant( guardianSigner guardiansigner.GuardianSigner, // the guardian signer used for signing observation requests gst *common.GuardianSetState, // used to get the current guardian set index when sending observation requests msgChan chan<- *common.MessagePublication, // the channel where transfers received by the accountant runnable should be published + submitObservationBatchSize int, // maximum number of observations to submit to the contract in one transaction env common.Environment, // Controls the set of token bridges to be monitored ) *Accountant { + if submitObservationBatchSize <= 0 { + submitObservationBatchSize = DefaultSubmitObservationBatchSize + } + return &Accountant{ - ctx: ctx, - logger: logger.With(zap.String("component", "gacct")), - db: db, - obsvReqWriteC: obsvReqWriteC, - contract: contract, - wsUrl: wsUrl, - wormchainConn: wormchainConn, - enforceFlag: enforceFlag, - guardianSigner: guardianSigner, - gst: gst, - guardianAddr: ethCrypto.PubkeyToAddress(guardianSigner.PublicKey(ctx)), - msgChan: msgChan, - tokenBridges: make(validEmitters), - pendingTransfers: make(map[string]*pendingEntry), - subChan: make(chan *common.MessagePublication, subChanSize), - env: env, + ctx: ctx, + logger: logger.With(zap.String("component", "gacct")), + db: db, + obsvReqWriteC: obsvReqWriteC, + contract: contract, + wsUrl: wsUrl, + wormchainConn: wormchainConn, + enforceFlag: enforceFlag, + guardianSigner: guardianSigner, + gst: gst, + guardianAddr: ethCrypto.PubkeyToAddress(guardianSigner.PublicKey(ctx)), + msgChan: msgChan, + tokenBridges: make(validEmitters), + pendingTransfers: make(map[string]*pendingEntry), + subChan: make(chan *common.MessagePublication, subChanSize), + submitObservationBatchSize: submitObservationBatchSize, + env: env, nttContract: nttContract, nttWormchainConn: nttWormchainConn, @@ -156,7 +163,7 @@ func NewAccountant( // Start initializes the accountant and starts the worker and watcher runnables. func (acct *Accountant) Start(ctx context.Context) error { - acct.logger.Debug("entering Start", zap.Bool("enforceFlag", acct.enforceFlag), zap.Bool("baseEnabled", acct.baseEnabled()), zap.Bool("nttEnabled", acct.nttEnabled())) + acct.logger.Debug("entering Start", zap.Bool("enforceFlag", acct.enforceFlag), zap.Bool("baseEnabled", acct.baseEnabled()), zap.Bool("nttEnabled", acct.nttEnabled()), zap.Int("submitObservationBatchSize", acct.submitObservationBatchSize)) acct.pendingTransfersLock.Lock() defer acct.pendingTransfersLock.Unlock() diff --git a/node/pkg/accountant/accountant_test.go b/node/pkg/accountant/accountant_test.go index 86a688153c8..0c31165d28e 100644 --- a/node/pkg/accountant/accountant_test.go +++ b/node/pkg/accountant/accountant_test.go @@ -153,6 +153,7 @@ func newAccountantForTest( guardianSigner, gst, acctWriteC, + DefaultSubmitObservationBatchSize, env, ) @@ -767,6 +768,7 @@ func newAccountantForAuditTest( guardianSigner, gst, acctWriteC, + DefaultSubmitObservationBatchSize, common.GoTest, // Use GoTest to avoid starting worker goroutines that consume from subChan ) diff --git a/node/pkg/accountant/submit_obs.go b/node/pkg/accountant/submit_obs.go index b688b839302..b1034fcde33 100644 --- a/node/pkg/accountant/submit_obs.go +++ b/node/pkg/accountant/submit_obs.go @@ -22,8 +22,8 @@ import ( ) const ( - batchSize = 100 // Observations per batch (limited by wasm contract input size of 64KB) - batchTimeout = 2 * time.Second // Time to collect observations before submitting + DefaultSubmitObservationBatchSize = 100 // Observations per batch (limited by wasm contract input size of 64KB) + batchTimeout = 2 * time.Second // Time to collect observations before submitting ) // baseWorker is the entry point for the base accountant worker. @@ -68,7 +68,7 @@ func (acct *Accountant) handleBatch(ctx context.Context, subChan chan *common.Me ctx, cancel := context.WithTimeout(ctx, batchTimeout) defer cancel() - msgs, err := common.ReadFromChannelWithTimeout[*common.MessagePublication](ctx, subChan, batchSize) + msgs, err := common.ReadFromChannelWithTimeout[*common.MessagePublication](ctx, subChan, acct.submitObservationBatchSize) if err != nil && !errors.Is(err, context.DeadlineExceeded) { return fmt.Errorf("failed to read messages from channel for %s: %w", tag, err) } diff --git a/node/pkg/node/node_test.go b/node/pkg/node/node_test.go index 92f0b573057..9eef7ac682a 100644 --- a/node/pkg/node/node_test.go +++ b/node/pkg/node/node_test.go @@ -926,10 +926,27 @@ func TestGuardianConfigs(t *testing.T) { nil, // wormchainConn "", // nttContract nil, // nttWormchainConn + 100, // submitObservationBatchSize ), }, err: ComponentDependencyError{componentName: "accountant", dependencyName: "db"}.Error(), }, + { + name: "invalid-accountant-submit-observation-batch-size", + opts: []*GuardianOption{ + GuardianOptionDatabase(nil), + GuardianOptionAccountant( + "", // websocket + "", // contract + false, // enforcing + nil, // wormchainConn + "", // nttContract + nil, // nttWormchainConn + 0, // submitObservationBatchSize + ), + }, + err: "accountantSubmitObservationBatchSize must be greater than zero", + }, { name: "double-configuration", opts: []*GuardianOption{ diff --git a/node/pkg/node/options.go b/node/pkg/node/options.go index d611eac550f..c4c2f7f074c 100644 --- a/node/pkg/node/options.go +++ b/node/pkg/node/options.go @@ -169,7 +169,7 @@ func GuardianOptionQueryHandler(ccqEnabled bool, allowedRequesters string) *Guar }} } -// GuardianOptionNoAccountant disables the accountant. It is a shorthand for GuardianOptionAccountant("", "", false, nil) +// GuardianOptionNoAccountant disables the accountant. // Dependencies: none func GuardianOptionNoAccountant() *GuardianOption { return &GuardianOption{ @@ -189,11 +189,16 @@ func GuardianOptionAccountant( wormchainConn *wormconn.ClientConn, nttContract string, nttWormchainConn *wormconn.ClientConn, + submitObservationBatchSize int, ) *GuardianOption { return &GuardianOption{ name: "accountant", dependencies: []string{"db"}, f: func(ctx context.Context, logger *zap.Logger, g *G) error { + if submitObservationBatchSize <= 0 { + return errors.New("accountantSubmitObservationBatchSize must be greater than zero") + } + // Set up the accountant. If the accountant smart contract is configured, we will instantiate the accountant and VAAs // will be passed to it for processing. It will forward all token bridge transfers to the accountant contract. // If accountantCheckEnabled is set to true, token bridge transfers will not be signed and published until they @@ -237,6 +242,7 @@ func GuardianOptionAccountant( g.guardianSigner, g.gst, g.acctC.writeC, + submitObservationBatchSize, g.env, ) From 113c69b63724d1c5d0334f8c8f338fddbbec9fcd Mon Sep 17 00:00:00 2001 From: Dirk Brink Date: Tue, 30 Jun 2026 17:28:20 -0600 Subject: [PATCH 6/6] Only run accountant audit for NTT/WTT when enabled --- node/pkg/accountant/accountant.go | 10 ++-- node/pkg/accountant/accountant_test.go | 71 ++++++++++++++++++++++++-- node/pkg/accountant/audit.go | 22 ++++---- 3 files changed, 88 insertions(+), 15 deletions(-) diff --git a/node/pkg/accountant/accountant.go b/node/pkg/accountant/accountant.go index 013af7e9c11..08d36cbb92b 100644 --- a/node/pkg/accountant/accountant.go +++ b/node/pkg/accountant/accountant.go @@ -225,9 +225,13 @@ func (acct *Accountant) Start(ctx context.Context) error { return fmt.Errorf("failed to start watcher: %w", err) } - if err := supervisor.Run(ctx, "acctaudit", common.WrapWithScissors(acct.audit, "acctaudit")); err != nil { - return fmt.Errorf("failed to start audit worker: %w", err) - } + } + } + + // Start the audit worker if not mocking/testing and either Global or NTT accountant are enabled + if acct.env != common.AccountantMock && acct.env != common.GoTest && (acct.baseEnabled() || acct.nttEnabled()) { + if err := supervisor.Run(ctx, "acctaudit", common.WrapWithScissors(acct.audit, "acctaudit")); err != nil { + return fmt.Errorf("failed to start audit worker: %w", err) } } diff --git a/node/pkg/accountant/accountant_test.go b/node/pkg/accountant/accountant_test.go index 0c31165d28e..7ea93a334c9 100644 --- a/node/pkg/accountant/accountant_test.go +++ b/node/pkg/accountant/accountant_test.go @@ -102,10 +102,19 @@ type AuditMockWormchainConn struct { allPendingTransfersErr error batchTransferStatusResp []byte batchTransferStatusErr error + + queryLock sync.Mutex + queries []string + contractAddresses []string } func (c *AuditMockWormchainConn) SubmitQuery(ctx context.Context, contractAddress string, query []byte) ([]byte, error) { queryStr := string(query) + c.queryLock.Lock() + c.queries = append(c.queries, queryStr) + c.contractAddresses = append(c.contractAddresses, contractAddress) + c.queryLock.Unlock() + if strings.Contains(queryStr, "all_pending_transfers") { return c.allPendingTransfersResp, c.allPendingTransfersErr } @@ -115,6 +124,12 @@ func (c *AuditMockWormchainConn) SubmitQuery(ctx context.Context, contractAddres return []byte{}, nil } +func (c *AuditMockWormchainConn) QueryCount() int { + c.queryLock.Lock() + defer c.queryLock.Unlock() + return len(c.queries) +} + func newAccountantForTest( t *testing.T, logger *zap.Logger, @@ -739,6 +754,20 @@ func newAccountantForAuditTest( obsvReqWriteC chan *gossipv1.ObservationRequest, acctWriteC chan *common.MessagePublication, wormchainConn *AuditMockWormchainConn, +) *Accountant { + return newAccountantForAuditModeTest(t, logger, ctx, obsvReqWriteC, acctWriteC, "0xdeadbeef", wormchainConn, "", nil) +} + +func newAccountantForAuditModeTest( + t *testing.T, + logger *zap.Logger, + ctx context.Context, + obsvReqWriteC chan *gossipv1.ObservationRequest, + acctWriteC chan *common.MessagePublication, + contract string, + wormchainConn *AuditMockWormchainConn, + nttContract string, + nttWormchainConn *AuditMockWormchainConn, ) *Accountant { var db guardianDB.MockAccountantDB @@ -759,12 +788,12 @@ func newAccountantForAuditTest( logger, &db, obsvReqWriteC, - "0xdeadbeef", + contract, "none", wormchainConn, true, // enforceFlag - "", - nil, + nttContract, + nttWormchainConn, guardianSigner, gst, acctWriteC, @@ -785,6 +814,42 @@ var ( testTxHash = []byte{0x06, 0xf5, 0x41, 0xf5, 0xec, 0xfc, 0x43, 0x40, 0x7c, 0x31, 0x58, 0x7a, 0xa6, 0xac, 0x3a, 0x68, 0x9e, 0x89, 0x60, 0xf3, 0x6d, 0xc2, 0x3c, 0x33, 0x2d, 0xb5, 0x51, 0x0d, 0xfc, 0x6a, 0x40, 0x64} ) +func TestRunAuditBaseOnlySkipsNttAudit(t *testing.T) { + ctx := context.Background() + logger := zaptest.NewLogger(t) + obsvReqWriteC := make(chan *gossipv1.ObservationRequest, 10) + acctChan := make(chan *common.MessagePublication, MsgChannelCapacity) + wormchainConn := &AuditMockWormchainConn{ + allPendingTransfersResp: []byte(`{"pending":[]}`), + } + + acct := newAccountantForAuditModeTest(t, logger, ctx, obsvReqWriteC, acctChan, "base-contract", wormchainConn, "", nil) + + require.NotPanics(t, func() { + acct.runAudit(ctx) + }) + assert.Equal(t, 1, wormchainConn.QueryCount(), "expected base audit query") + assert.Equal(t, []string{"base-contract"}, wormchainConn.contractAddresses) +} + +func TestRunAuditNttOnlySkipsBaseAudit(t *testing.T) { + ctx := context.Background() + logger := zaptest.NewLogger(t) + obsvReqWriteC := make(chan *gossipv1.ObservationRequest, 10) + acctChan := make(chan *common.MessagePublication, MsgChannelCapacity) + nttWormchainConn := &AuditMockWormchainConn{ + allPendingTransfersResp: []byte(`{"pending":[]}`), + } + + acct := newAccountantForAuditModeTest(t, logger, ctx, obsvReqWriteC, acctChan, "", nil, "ntt-contract", nttWormchainConn) + + require.NotPanics(t, func() { + acct.runAudit(ctx) + }) + assert.Equal(t, 1, nttWormchainConn.QueryCount(), "expected NTT audit query") + assert.Equal(t, []string{"ntt-contract"}, nttWormchainConn.contractAddresses) +} + func TestPerformAuditResubmitsUnsignedTransfer(t *testing.T) { ctx := context.Background() logger := zaptest.NewLogger(t) diff --git a/node/pkg/accountant/audit.go b/node/pkg/accountant/audit.go index 32b87cf173e..8d0ccd2af57 100644 --- a/node/pkg/accountant/audit.go +++ b/node/pkg/accountant/audit.go @@ -155,15 +155,19 @@ func (acct *Accountant) audit(ctx context.Context) error { // runAudit is the entry point for the audit of the pending transfer map. It creates a temporary map of all pending transfers and invokes the main audit function. func (acct *Accountant) runAudit(ctx context.Context) { - knownPendingTransferMap := acct.createAuditMap(false) - acct.logger.Debug("in AuditPendingTransfers: starting base audit", zap.Int("numPending", numPendingEntries(knownPendingTransferMap))) - acct.performAudit(ctx, knownPendingTransferMap, acct.wormchainConn, acct.contract) - acct.logger.Debug("in AuditPendingTransfers: finished base audit") - - knownPendingNttTransferMap := acct.createAuditMap(true) - acct.logger.Debug("in AuditPendingTransfers: starting ntt audit", zap.Int("numPending", numPendingEntries(knownPendingNttTransferMap))) - acct.performAudit(ctx, knownPendingNttTransferMap, acct.nttWormchainConn, acct.nttContract) - acct.logger.Debug("in AuditPendingTransfers: finished ntt audit") + if acct.baseEnabled() { + knownPendingTransferMap := acct.createAuditMap(false) + acct.logger.Debug("in AuditPendingTransfers: starting base audit", zap.Int("numPending", numPendingEntries(knownPendingTransferMap))) + acct.performAudit(ctx, knownPendingTransferMap, acct.wormchainConn, acct.contract) + acct.logger.Debug("in AuditPendingTransfers: finished base audit") + } + + if acct.nttEnabled() { + knownPendingNttTransferMap := acct.createAuditMap(true) + acct.logger.Debug("in AuditPendingTransfers: starting ntt audit", zap.Int("numPending", numPendingEntries(knownPendingNttTransferMap))) + acct.performAudit(ctx, knownPendingNttTransferMap, acct.nttWormchainConn, acct.nttContract) + acct.logger.Debug("in AuditPendingTransfers: finished ntt audit") + } } // numPendingEntries returns the total number of non-nil pending entries across all keys