From d9d965dcba7846a50f42f5bd243b018718f18f36 Mon Sep 17 00:00:00 2001 From: Dirk Brink Date: Mon, 17 Aug 2026 14:51:50 -0600 Subject: [PATCH 1/2] node: Update accountant audit logic --- node/pkg/accountant/metrics.go | 10 ++ node/pkg/accountant/submit_obs.go | 118 ++++++++++++++-- node/pkg/accountant/submit_obs_test.go | 179 +++++++++++++++++++++++++ 3 files changed, 295 insertions(+), 12 deletions(-) diff --git a/node/pkg/accountant/metrics.go b/node/pkg/accountant/metrics.go index ac17496b471..09ff2032a8c 100644 --- a/node/pkg/accountant/metrics.go +++ b/node/pkg/accountant/metrics.go @@ -62,4 +62,14 @@ var ( Name: "global_accountant_channel_submit_timeouts", Help: "Total number of channel submit timeouts during audit", }) + batchSizeSplits = promauto.NewCounter( + prometheus.CounterOpts{ + Name: "global_accountant_batch_size_splits_total", + Help: "Total number of times a batch of observations was split into multiple batches to stay within the transaction size limit", + }) + oversizedObservations = promauto.NewCounter( + prometheus.CounterOpts{ + Name: "global_accountant_oversized_observations_total", + Help: "Total number of observations that exceed the transaction size limit even in a batch by themselves", + }) ) diff --git a/node/pkg/accountant/submit_obs.go b/node/pkg/accountant/submit_obs.go index b1034fcde33..a04ab161bce 100644 --- a/node/pkg/accountant/submit_obs.go +++ b/node/pkg/accountant/submit_obs.go @@ -2,6 +2,7 @@ package accountant import ( "context" + "encoding/base64" "encoding/hex" "encoding/json" "errors" @@ -22,7 +23,8 @@ import ( ) const ( - DefaultSubmitObservationBatchSize = 100 // Observations per batch (limited by wasm contract input size of 64KB) + DefaultSubmitObservationBatchSize = 100 // Maximum observations per batch, also subject to maxSubmitObservationsMsgSize + maxSubmitObservationsMsgSize = 64 * 1024 // Maximum size of the marshaled submit_observations message (the wasm contract input is limited to 64KB) batchTimeout = 2 * time.Second // Time to collect observations before submitting ) @@ -95,11 +97,98 @@ func (acct *Accountant) handleBatch(ctx context.Context, subChan chan *common.Me return fmt.Errorf("guardian index greater than max uint32 %v", guardianIndex) } - acct.submitObservationsToContract(msgs, gs.Index, uint32(guardianIndex), wormchainConn, contract, prefix, tag) // #nosec G115 -- This is checked above + batches, oversized := packObservationBatches(msgs, maxSubmitObservationsMsgSize) + if len(batches) > 1 { + acct.logger.Info(fmt.Sprintf("split observations for %s into multiple batches to stay within the transaction size limit", tag), zap.Int("numMsgs", len(msgs)), zap.Int("numBatches", len(batches))) + batchSizeSplits.Inc() + } + for _, msg := range oversized { + // The message is submitted anyway (alone, so it can't take any other observations down with it) in case our size accounting is too conservative. + acct.logger.Error(fmt.Sprintf("observation exceeds the transaction size limit for %s even in a batch by itself", tag), zap.String("msgId", msg.MessageIDString()), zap.Int("payloadLen", len(msg.Payload))) + oversizedObservations.Inc() + } + + for _, batch := range batches { + acct.submitObservationsToContract(batch, gs.Index, uint32(guardianIndex), wormchainConn, contract, prefix, tag) // #nosec G115 -- This is checked above + } transfersSubmitted.Add(float64(len(msgs))) return nil } +// packObservationBatches partitions the messages into batches whose marshaled submit_observations messages each stay within +// maxMsgSize. Each message is placed in the first batch with enough room for it, so a large message that does not fit in the +// current batch is deferred to a later batch rather than failing the messages around it. A message too large to fit even in a +// batch by itself is returned in oversized as well as being placed in its own batch. +func packObservationBatches(msgs []*common.MessagePublication, maxMsgSize int) (batches [][]*common.MessagePublication, oversized []*common.MessagePublication) { + // batchSizes[i] is the size the marshaled observations array for batches[i] would have, including the enclosing brackets. + var batchSizes []int + for _, msg := range msgs { + obsSize := marshaledObservationSize(msg) + placed := false + for idx := range batches { + // Adding an observation to a batch grows the observations array by the observation plus a separating comma. + if len(batches[idx]) < DefaultSubmitObservationBatchSize && submitObservationsMsgSize(batchSizes[idx]+obsSize+jsonCommaSize) <= maxMsgSize { + batches[idx] = append(batches[idx], msg) + batchSizes[idx] += obsSize + jsonCommaSize + placed = true + break + } + } + if !placed { + if submitObservationsMsgSize(obsSize+jsonBracketsSize) > maxMsgSize { + oversized = append(oversized, msg) + } + batches = append(batches, []*common.MessagePublication{msg}) + batchSizes = append(batchSizes, obsSize+jsonBracketsSize) + } + } + return batches, oversized +} + +// marshaledObservationSize returns the number of bytes the observation for the message occupies in the marshaled observations array. +func marshaledObservationSize(msg *common.MessagePublication) int { + bytes, err := json.Marshal(makeObservation(msg)) + if err != nil { + // Marshaling an Observation cannot fail, since none of its field types can produce a marshaling error. + panic(fmt.Sprintf("failed to marshal observation: %v", err)) + } + return len(bytes) +} + +// Sizes of the JSON punctuation accounted for when computing how large a marshaled submit_observations message will be. +const ( + jsonQuotesSize = len(`""`) // The quotes around a JSON string, such as the base64-encoded observations + jsonBracketsSize = len(`[]`) // The brackets around a JSON array, such as the marshaled observations + jsonCommaSize = len(`,`) // The comma between JSON array elements +) + +// submitObservationsMsgSize returns the size of the marshaled submit_observations message for an observations array of +// obsArraySize bytes, assuming worst-case sizes for the other fields. +func submitObservationsMsgSize(obsArraySize int) int { + // The marshaled observations array appears in the message as a quoted base64 string. + return submitObservationsMsgOverhead + jsonQuotesSize + base64.StdEncoding.EncodedLen(obsArraySize) +} + +// submitObservationsMsgOverhead is the number of bytes in a marshaled submit_observations message excluding the base64-encoded +// observations string, computed with worst-case values for the other fields. +var submitObservationsMsgOverhead = func() int { + sig := make(SignatureBytes, 65) //nolint:mnd // Length of an ECDSA (r, s, v) signature + for idx := range sig { + sig[idx] = math.MaxUint8 + } + bytes, err := json.Marshal(SubmitObservationsMsg{ + Params: SubmitObservationsParams{ + Observations: []byte{}, + GuardianSetIndex: math.MaxUint32, + Signature: SignatureType{Index: math.MaxUint32, Signature: sig}, + }, + }) + if err != nil { + panic(fmt.Sprintf("failed to marshal submit observations message: %v", err)) + } + return len(bytes) - jsonQuotesSize // Marshaling the empty observations produces just the quotes of an empty string. +}() + // removeCompleted drops any messages that are no longer in the pending transfer map. This is to handle the case where the contract reports // that a transfer is committed while it is in the channel. There is no point in submitting the observation once the transfer is committed. func (acct *Accountant) removeCompleted(msgs []*common.MessagePublication) []*common.MessagePublication { @@ -188,6 +277,20 @@ type ( var SubmitObservationPrefix = []byte("acct_sub_obsfig_000000000000000000|") var NttSubmitObservationPrefix = []byte("ntt_acct_sub_obsfig_00000000000000|") +// makeObservation converts a message publication into the observation that is submitted to the smart contract. +func makeObservation(msg *common.MessagePublication) Observation { + return Observation{ + TxHash: msg.TxID, + Timestamp: uint32(msg.Timestamp.Unix()), // #nosec G115 -- This conversion is safe until year 2106 + Nonce: msg.Nonce, + EmitterChain: uint16(msg.EmitterChain), + EmitterAddress: msg.EmitterAddress, + Sequence: msg.Sequence, + ConsistencyLevel: msg.ConsistencyLevel, + Payload: msg.Payload, + } +} + func (k TransferKey) String() string { return fmt.Sprintf("%v/%v/%v", k.EmitterChain, hex.EncodeToString(k.EmitterAddress[:]), k.Sequence) } @@ -315,16 +418,7 @@ func SubmitObservationsToContract( ) (*sdktx.BroadcastTxResponse, error) { obs := make([]Observation, len(msgs)) for idx, msg := range msgs { - obs[idx] = Observation{ - TxHash: msg.TxID, - Timestamp: uint32(msg.Timestamp.Unix()), // #nosec G115 -- This conversion is safe until year 2106 - Nonce: msg.Nonce, - EmitterChain: uint16(msg.EmitterChain), - EmitterAddress: msg.EmitterAddress, - Sequence: msg.Sequence, - ConsistencyLevel: msg.ConsistencyLevel, - Payload: msg.Payload, - } + obs[idx] = makeObservation(msg) logger.Debug("in SubmitObservationsToContract, encoding observation", zap.String("contract", contract), diff --git a/node/pkg/accountant/submit_obs_test.go b/node/pkg/accountant/submit_obs_test.go index 83a5753abec..4c7c692cb52 100644 --- a/node/pkg/accountant/submit_obs_test.go +++ b/node/pkg/accountant/submit_obs_test.go @@ -3,8 +3,12 @@ package accountant import ( // "encoding/hex" "encoding/json" + "math" + "slices" "testing" + "time" + "github.com/certusone/wormhole/node/pkg/common" "github.com/wormhole-foundation/wormhole/sdk/vaa" "github.com/stretchr/testify/assert" @@ -65,3 +69,178 @@ func TestParseObservationResponseData(t *testing.T) { assert.Equal(t, expectedResult0, responses[0]) assert.Equal(t, expectedResult1, responses[1]) } + +func makeMsgForPackingTest(t *testing.T, sequence uint64, payloadLen int) *common.MessagePublication { + t.Helper() + emitterAddr, err := vaa.StringToAddress("0x0290fb167208af455bb137780163b7b7a9a10c16") + require.NoError(t, err) + return &common.MessagePublication{ + TxID: []byte("0123456789abcdef0123456789abcdef"), + Timestamp: time.Unix(1654543099, 0), + Nonce: 123456, + Sequence: sequence, + EmitterChain: vaa.ChainIDEthereum, + EmitterAddress: emitterAddr, + Payload: make([]byte, payloadLen), + } +} + +// repeatedPayloadLens returns num copies of payloadLen, for building a batch of identical messages. +func repeatedPayloadLens(payloadLen int, num int) []int { + lens := make([]int, num) + for i := range lens { + lens[i] = payloadLen + } + return lens +} + +// indexRange returns the indices from start (inclusive) to end (exclusive). +func indexRange(start int, end int) []int { + indices := make([]int, 0, end-start) + for i := start; i < end; i++ { + indices = append(indices, i) + } + return indices +} + +// marshaledMsgSizeForBatch builds the submit_observations message for a batch the same way SubmitObservationsToContract does, +// with worst-case values for the fields other than the observations, and returns its marshaled size. +func marshaledMsgSizeForBatch(t *testing.T, batch []*common.MessagePublication) int { + t.Helper() + obs := make([]Observation, len(batch)) + for idx, msg := range batch { + obs[idx] = makeObservation(msg) + } + obsBytes, err := json.Marshal(obs) + require.NoError(t, err) + + sig := make(SignatureBytes, 65) + for idx := range sig { + sig[idx] = math.MaxUint8 + } + msgBytes, err := json.Marshal(SubmitObservationsMsg{ + Params: SubmitObservationsParams{ + Observations: obsBytes, + GuardianSetIndex: math.MaxUint32, + Signature: SignatureType{Index: math.MaxUint32, Signature: sig}, + }, + }) + require.NoError(t, err) + return len(msgBytes) +} + +func TestSubmitObservationsMsgSizeMatchesMarshaledMsg(t *testing.T) { + // The size the packing logic computes for a batch should exactly match the size of the real marshaled message. + batch := []*common.MessagePublication{ + makeMsgForPackingTest(t, 1, 0), + makeMsgForPackingTest(t, 12345678, 100), + makeMsgForPackingTest(t, math.MaxUint64, 4000), + } + obs := make([]Observation, len(batch)) + for idx, msg := range batch { + obs[idx] = makeObservation(msg) + } + obsBytes, err := json.Marshal(obs) + require.NoError(t, err) + + assert.Equal(t, marshaledMsgSizeForBatch(t, batch), submitObservationsMsgSize(len(obsBytes))) +} + +func TestPackObservationBatches(t *testing.T) { + testCases := []struct { + name string + payloadLens []int // One message per entry; the message's sequence number is its index. + + // maxMsgSize returns the batch size limit for the test, given the test messages. Nil means the production limit. + maxMsgSize func(msgs []*common.MessagePublication) int + + // expBatches are the expected batches as indices into the messages. Nil means the batch composition is not asserted, + // only the shared invariants (used when the composition would depend on hand-computed observation sizes). + expBatches [][]int + expOversized []int + minBatches int + }{ + { + name: "keeps small messages together in one batch", + payloadLens: repeatedPayloadLens(100, 100), + expBatches: [][]int{indexRange(0, 100)}, + }, + { + name: "respects count limit", + payloadLens: repeatedPayloadLens(100, 205), + expBatches: [][]int{indexRange(0, 100), indexRange(100, 200), indexRange(200, 205)}, + }, + { + name: "splits on size", + payloadLens: repeatedPayloadLens(10*1024, 10), + minBatches: 2, + }, + { + name: "defers large message to a later batch", + payloadLens: []int{50, 2000, 50}, + maxMsgSize: func(msgs []*common.MessagePublication) int { + // Fits the large message on its own (with a little headroom), but not together with a small one. + return submitObservationsMsgSize(marshaledObservationSize(msgs[1]) + 10) + }, + expBatches: [][]int{{0, 2}, {1}}, + }, + { + name: "isolates oversized message in its own batch", + payloadLens: []int{50, 5000, 50}, + maxMsgSize: func(msgs []*common.MessagePublication) int { + // Cannot fit the huge message even on its own. + return submitObservationsMsgSize(marshaledObservationSize(msgs[1]) - 10) + }, + expBatches: [][]int{{0, 2}, {1}}, + expOversized: []int{1}, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + msgs := make([]*common.MessagePublication, len(tc.payloadLens)) + for idx, payloadLen := range tc.payloadLens { + msgs[idx] = makeMsgForPackingTest(t, uint64(idx), payloadLen) // #nosec G115 -- test values are small + } + maxMsgSize := maxSubmitObservationsMsgSize + if tc.maxMsgSize != nil { + maxMsgSize = tc.maxMsgSize(msgs) + } + expOversized := make([]*common.MessagePublication, 0, len(tc.expOversized)) + for _, idx := range tc.expOversized { + expOversized = append(expOversized, msgs[idx]) + } + + batches, oversized := packObservationBatches(msgs, maxMsgSize) + + assert.ElementsMatch(t, expOversized, oversized) + assert.GreaterOrEqual(t, len(batches), tc.minBatches) + + // Shared invariants: every message lands in exactly one batch, and every batch respects the count limit and, + // unless it holds an oversized message (which always gets a batch to itself), the size limit. + var packed []*common.MessagePublication + for _, batch := range batches { + require.NotEmpty(t, batch) + assert.LessOrEqual(t, len(batch), DefaultSubmitObservationBatchSize) + if !slices.Contains(expOversized, batch[0]) { + assert.LessOrEqual(t, marshaledMsgSizeForBatch(t, batch), maxMsgSize) + } else { + require.Len(t, batch, 1) + } + packed = append(packed, batch...) + } + assert.ElementsMatch(t, msgs, packed) + + if tc.expBatches != nil { + require.Len(t, batches, len(tc.expBatches)) + for batchIdx, expIndices := range tc.expBatches { + expBatch := make([]*common.MessagePublication, 0, len(expIndices)) + for _, idx := range expIndices { + expBatch = append(expBatch, msgs[idx]) + } + assert.Equal(t, expBatch, batches[batchIdx], "batch %d", batchIdx) + } + } + }) + } +} From 81485f3734344a68a6d98938ea7b707b6233c508 Mon Sep 17 00:00:00 2001 From: Dirk Brink Date: Mon, 17 Aug 2026 15:39:28 -0600 Subject: [PATCH 2/2] Claude PR feedback --- node/pkg/accountant/submit_obs.go | 15 ++++++++------- node/pkg/accountant/submit_obs_test.go | 22 ++++++++++++++++++++-- 2 files changed, 28 insertions(+), 9 deletions(-) diff --git a/node/pkg/accountant/submit_obs.go b/node/pkg/accountant/submit_obs.go index a04ab161bce..afe402bb211 100644 --- a/node/pkg/accountant/submit_obs.go +++ b/node/pkg/accountant/submit_obs.go @@ -97,7 +97,7 @@ func (acct *Accountant) handleBatch(ctx context.Context, subChan chan *common.Me return fmt.Errorf("guardian index greater than max uint32 %v", guardianIndex) } - batches, oversized := packObservationBatches(msgs, maxSubmitObservationsMsgSize) + batches, oversized := packObservationBatches(msgs, acct.submitObservationBatchSize, maxSubmitObservationsMsgSize) if len(batches) > 1 { acct.logger.Info(fmt.Sprintf("split observations for %s into multiple batches to stay within the transaction size limit", tag), zap.Int("numMsgs", len(msgs)), zap.Int("numBatches", len(batches))) batchSizeSplits.Inc() @@ -115,11 +115,12 @@ func (acct *Accountant) handleBatch(ctx context.Context, subChan chan *common.Me return nil } -// packObservationBatches partitions the messages into batches whose marshaled submit_observations messages each stay within -// maxMsgSize. Each message is placed in the first batch with enough room for it, so a large message that does not fit in the -// current batch is deferred to a later batch rather than failing the messages around it. A message too large to fit even in a -// batch by itself is returned in oversized as well as being placed in its own batch. -func packObservationBatches(msgs []*common.MessagePublication, maxMsgSize int) (batches [][]*common.MessagePublication, oversized []*common.MessagePublication) { +// packObservationBatches partitions the messages into batches of at most maxBatchCount messages whose marshaled +// submit_observations messages each stay within maxMsgSize. Each message is placed in the first batch with enough room for it, +// so a large message that does not fit in the current batch is deferred to a later batch rather than failing the messages +// around it. A message too large to fit even in a batch by itself is returned in oversized as well as being placed in its own +// batch. +func packObservationBatches(msgs []*common.MessagePublication, maxBatchCount int, maxMsgSize int) (batches [][]*common.MessagePublication, oversized []*common.MessagePublication) { // batchSizes[i] is the size the marshaled observations array for batches[i] would have, including the enclosing brackets. var batchSizes []int for _, msg := range msgs { @@ -127,7 +128,7 @@ func packObservationBatches(msgs []*common.MessagePublication, maxMsgSize int) ( placed := false for idx := range batches { // Adding an observation to a batch grows the observations array by the observation plus a separating comma. - if len(batches[idx]) < DefaultSubmitObservationBatchSize && submitObservationsMsgSize(batchSizes[idx]+obsSize+jsonCommaSize) <= maxMsgSize { + if len(batches[idx]) < maxBatchCount && submitObservationsMsgSize(batchSizes[idx]+obsSize+jsonCommaSize) <= maxMsgSize { batches[idx] = append(batches[idx], msg) batchSizes[idx] += obsSize + jsonCommaSize placed = true diff --git a/node/pkg/accountant/submit_obs_test.go b/node/pkg/accountant/submit_obs_test.go index 4c7c692cb52..d29bed26e53 100644 --- a/node/pkg/accountant/submit_obs_test.go +++ b/node/pkg/accountant/submit_obs_test.go @@ -151,6 +151,9 @@ func TestPackObservationBatches(t *testing.T) { name string payloadLens []int // One message per entry; the message's sequence number is its index. + // maxBatchCount is the batch count limit for the test. Zero means the default limit. + maxBatchCount int + // maxMsgSize returns the batch size limit for the test, given the test messages. Nil means the production limit. maxMsgSize func(msgs []*common.MessagePublication) int @@ -160,6 +163,11 @@ func TestPackObservationBatches(t *testing.T) { expOversized []int minBatches int }{ + { + name: "returns no batches for no messages", + payloadLens: []int{}, + expBatches: [][]int{}, + }, { name: "keeps small messages together in one batch", payloadLens: repeatedPayloadLens(100, 100), @@ -170,6 +178,12 @@ func TestPackObservationBatches(t *testing.T) { payloadLens: repeatedPayloadLens(100, 205), expBatches: [][]int{indexRange(0, 100), indexRange(100, 200), indexRange(200, 205)}, }, + { + name: "respects configured count limit", + payloadLens: repeatedPayloadLens(100, 5), + maxBatchCount: 2, + expBatches: [][]int{{0, 1}, {2, 3}, {4}}, + }, { name: "splits on size", payloadLens: repeatedPayloadLens(10*1024, 10), @@ -202,6 +216,10 @@ func TestPackObservationBatches(t *testing.T) { for idx, payloadLen := range tc.payloadLens { msgs[idx] = makeMsgForPackingTest(t, uint64(idx), payloadLen) // #nosec G115 -- test values are small } + maxBatchCount := tc.maxBatchCount + if maxBatchCount == 0 { + maxBatchCount = DefaultSubmitObservationBatchSize + } maxMsgSize := maxSubmitObservationsMsgSize if tc.maxMsgSize != nil { maxMsgSize = tc.maxMsgSize(msgs) @@ -211,7 +229,7 @@ func TestPackObservationBatches(t *testing.T) { expOversized = append(expOversized, msgs[idx]) } - batches, oversized := packObservationBatches(msgs, maxMsgSize) + batches, oversized := packObservationBatches(msgs, maxBatchCount, maxMsgSize) assert.ElementsMatch(t, expOversized, oversized) assert.GreaterOrEqual(t, len(batches), tc.minBatches) @@ -221,7 +239,7 @@ func TestPackObservationBatches(t *testing.T) { var packed []*common.MessagePublication for _, batch := range batches { require.NotEmpty(t, batch) - assert.LessOrEqual(t, len(batch), DefaultSubmitObservationBatchSize) + assert.LessOrEqual(t, len(batch), maxBatchCount) if !slices.Contains(expOversized, batch[0]) { assert.LessOrEqual(t, marshaledMsgSizeForBatch(t, batch), maxMsgSize) } else {