Cut process-stage CPU and allocations across the indexer and processors

go.mod

@@ -14,7 +14,6 @@ require (
 	github.com/go-playground/validator/v10 v10.27.0
 	github.com/golang-jwt/jwt/v5 v5.2.3
 	github.com/google/uuid v1.6.0
-	github.com/guregu/null v4.0.0+incompatible
 	github.com/jackc/pgx/v5 v5.7.6
 	github.com/lib/pq v1.10.9
 	github.com/pkg/errors v0.9.1

internal/data/accounts.go

@@ -95,6 +95,10 @@ func batchCopyAccounts[T any](
 
 	// COPY the link table using pgx binary format with native pgtype types. Upstream
 	// participants handling ensures that account address is not NULL here.
+	// Participants are deduplicated per parent row upstream, so a busy account
+	// repeats once per transaction/operation here; the memo collapses that to
+	// one decode per unique address per batch.
+	memo := make(types.AddressByteaMemo)
 	var rows [][]any
 	for id, addresses := range addressesByID {
 		ledgerCreatedAt, ok := ledgerCreatedAtByID[id]
@@ -107,7 +111,7 @@ func batchCopyAccounts[T any](
 		ledgerCreatedAtPgtype := pgtype.Timestamptz{Time: ledgerCreatedAt, Valid: true}
 		idPgtype := pgtype.Int8{Int64: id, Valid: true}
 		for addr := range addresses {
-			addrBytes, addrErr := types.AddressBytea(addr).Value()
+			addrBytes, addrErr := memo.Bytes(types.AddressBytea(addr))
 			if addrErr != nil {
 				return fmt.Errorf("converting address %s to bytes: %w", addr, addrErr)
 			}

internal/data/query_utils.go

@@ -74,21 +74,6 @@ func pgtypeInt2FromNullInt16(ni sql.NullInt16) pgtype.Int2 {
 	return pgtype.Int2{Int16: ni.Int16, Valid: ni.Valid}
 }
 
-// pgtypeBytesFromNullAddressBytea converts NullAddressBytea to bytes for BYTEA insert.
-func pgtypeBytesFromNullAddressBytea(na types.NullAddressBytea) ([]byte, error) {
-	if !na.Valid {
-		return nil, nil
-	}
-	val, err := na.Value()
-	if err != nil {
-		return nil, fmt.Errorf("converting address to bytes: %w", err)
-	}
-	if val == nil {
-		return nil, nil
-	}
-	return val.([]byte), nil
-}
-
 // CursorColumn represents a column name and its cursor value for decomposed pagination.
 type CursorColumn struct {
 	Name  string

internal/data/statechanges.go

@@ -154,6 +154,9 @@ func (m *StateChangeModel) BatchCopy(
 
 	start := time.Now()
 
+	// One memo shared by every row of this COPY, used only from pgx's single row-callback goroutine.
+	memo := make(types.AddressByteaMemo)
+
 	// COPY state_changes using pgx binary format with native pgtype types
 	copyCount, err := pgxTx.CopyFrom(
 		ctx,
@@ -174,29 +177,29 @@ func (m *StateChangeModel) BatchCopy(
 			sc := stateChanges[i]
 
 			// Convert account_id to BYTEA (required field)
-			accountBytes, err := sc.AccountID.Value()
+			accountBytes, err := memo.Bytes(sc.AccountID)
 			if err != nil {
 				return nil, fmt.Errorf("converting account_id: %w", err)
 			}
 
 			// Convert nullable account_id fields to BYTEA
-			signerBytes, err := pgtypeBytesFromNullAddressBytea(sc.SignerAccountID)
+			signerBytes, err := memo.NullBytes(sc.SignerAccountID)
 			if err != nil {
 				return nil, fmt.Errorf("converting signer_account_id: %w", err)
 			}
-			spenderBytes, err := pgtypeBytesFromNullAddressBytea(sc.SpenderAccountID)
+			spenderBytes, err := memo.NullBytes(sc.SpenderAccountID)
 			if err != nil {
 				return nil, fmt.Errorf("converting spender_account_id: %w", err)
 			}
-			creatorBytes, err := pgtypeBytesFromNullAddressBytea(sc.CreatorAccountID)
+			creatorBytes, err := memo.NullBytes(sc.CreatorAccountID)
 			if err != nil {
 				return nil, fmt.Errorf("converting creator_account_id: %w", err)
 			}
-			destinationBytes, err := pgtypeBytesFromNullAddressBytea(sc.DestinationAccountID)
+			destinationBytes, err := memo.NullBytes(sc.DestinationAccountID)
 			if err != nil {
 				return nil, fmt.Errorf("converting destination_account_id: %w", err)
 			}
-			tokenBytes, err := pgtypeBytesFromNullAddressBytea(sc.TokenID)
+			tokenBytes, err := memo.NullBytes(sc.TokenID)
 			if err != nil {
 				return nil, fmt.Errorf("converting token_id: %w", err)
 			}