node: Update accountant audit logic

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",
+		})
 )

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,99 @@ 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, 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()
+	}
+	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 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 {
+		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]) < maxBatchCount && 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 +278,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 +419,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),

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,196 @@ 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.
+
+		// 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
+
+		// 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:        "returns no batches for no messages",
+			payloadLens: []int{},
+			expBatches:  [][]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:          "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),
+			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
+			}
+			maxBatchCount := tc.maxBatchCount
+			if maxBatchCount == 0 {
+				maxBatchCount = DefaultSubmitObservationBatchSize
+			}
+			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, maxBatchCount, 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), maxBatchCount)
+				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)
+				}
+			}
+		})
+	}
+}