likenft-indexer: batch BookNFT eth_getLogs calls to reduce Alchemy CU spend

Batch per-contract acquire-book-nft-events tasks into groups of 50
(configurable via TASK_ACQUIRE_BOOKNFT_BATCH_SIZE). The inner handler
already supports multiple addresses, so this change is purely in the
task wrapping and enqueuing layer.

- Add TaskAcquireBookNFTBatchSize config field
- Change lifecycle task payload to accept []string with backward compat
- Batch lifecycles before enqueuing, one asynq task per batch
- Enqueue inside WithEnqueueing callback to avoid orphaned lifecycles
- Guard against negative fetch count when queue is full
- Add per-address fallback when batched eth_getLogs exceeds response limits

Author: williamchong

likenft-indexer/cmd/worker/config/config.go

@@ -27,6 +27,7 @@ type EnvConfig struct {
 	EvmEventQueryToBlockPadding            uint64 `envconfig:"EVM_EVENT_QUERY_TO_BLOCK_PADDING" default:"10"`
 
 	TaskAcquireBookNFTMaxQueueLength int `envconfig:"TASK_ACQUIRE_BOOKNFT_MAX_QUEUE_LENGTH" default:"500"`
+	TaskAcquireBookNFTBatchSize      int `envconfig:"TASK_ACQUIRE_BOOKNFT_BATCH_SIZE" default:"50"`
 
 	// The block height weight is the multiplier for the block height to be added to the score
 	TaskAcquireBookNFTNextProcessingBlockHeightWeight float64 `envconfig:"TASK_ACQUIRE_BOOKNFT_NEXT_PROCESSING_BLOCK_HEIGHT_WEIGHT" default:"0.00000001"`

likenft-indexer/cmd/worker/task/acquire_book_nft_events.go

@@ -5,8 +5,11 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"log/slog"
+	"strings"
 
 	appcontext "likenft-indexer/cmd/worker/context"
+	"likenft-indexer/cmd/worker/config"
 	"likenft-indexer/ent"
 	"likenft-indexer/ent/schema/typeutil"
 	"likenft-indexer/internal/database"
@@ -104,6 +107,22 @@ func HandleAcquireBookNFTEventsTask(ctx context.Context, t *asynq.Task) error {
 			cfg.EvmEventQueryNumberOfBlocksLimit,
 		)
 
+		if err != nil && len(addresses) > 1 {
+			var errCannotConvertLog *contractevmeventacquirer.ErrCannotConvertLog
+			if isResponseTooLargeError(err) || errors.As(err, &errCannotConvertLog) {
+				mylogger.Warn("batched acquire failed, falling back to per-address queries",
+					"addressCount", len(addresses), "err", err)
+				if fallbackErr := acquirePerAddress(
+					ctx, logger, cfg, evmEventQueryClient, evmEventRepository,
+					evmEventQueryClient, evmClient, nftClassRepository,
+					addresses, uint64(latestEventsBlockHeight),
+				); fallbackErr != nil {
+					return fallbackErr
+				}
+				continue
+			}
+		}
+
 		if err != nil {
 			mylogger.Error("acquirer.Acquire", "err", err)
 			var errCannotConvertLog *contractevmeventacquirer.ErrCannotConvertLog
@@ -130,6 +149,74 @@ func HandleAcquireBookNFTEventsTask(ctx context.Context, t *asynq.Task) error {
 	return nil
 }
 
+// isResponseTooLargeError checks if an RPC error indicates the eth_getLogs
+// response exceeded the provider's size limit.
+// Known error messages:
+//   - Alchemy: "Log response size exceeded."
+//   - Geth/others: "query returned more than 10000 results"
+func isResponseTooLargeError(err error) bool {
+	msg := strings.ToLower(err.Error())
+	return strings.Contains(msg, "log response size exceeded") ||
+		strings.Contains(msg, "query returned more than")
+}
+
+// acquirePerAddress retries event acquisition one address at a time.
+// Used as a fallback when a batched FilterLogs call exceeds response limits.
+func acquirePerAddress(
+	ctx context.Context,
+	logger *slog.Logger,
+	cfg *config.EnvConfig,
+	abiManager contractevmeventacquirer.ABIManager,
+	evmEventRepository database.EVMEventRepository,
+	evmEventQueryClient contractevmeventacquirer.EvmEventQueryClient,
+	evmClient contractevmeventacquirer.EvmClient,
+	nftClassRepository database.NFTClassRepository,
+	addresses []string,
+	fromBlock uint64,
+) error {
+	mylogger := logger.WithGroup("acquirePerAddress")
+
+	for _, addr := range addresses {
+		singleAcquirer := contractevmeventacquirer.NewContractEvmEventsAcquirer(
+			abiManager,
+			evmEventRepository,
+			evmEventQueryClient,
+			evmClient,
+			cfg.EvmEventQueryToBlockPadding,
+			contractevmeventacquirer.ContractEvmEventsAcquirerContractTypeBookNFT,
+			[]string{addr},
+		)
+
+		newBlockHeight, _, err := singleAcquirer.Acquire(
+			ctx,
+			logger,
+			fromBlock,
+			cfg.EvmEventQueryNumberOfBlocksLimit,
+		)
+		if err != nil {
+			mylogger.Error("acquirer.Acquire", "addr", addr, "err", err)
+			var errCannotConvertLog *contractevmeventacquirer.ErrCannotConvertLog
+			if errors.As(err, &errCannotConvertLog) {
+				if disableErr := nftClassRepository.DisableForIndexing(
+					ctx, errCannotConvertLog.Log.Address.Hex(), err.Error(),
+				); disableErr != nil {
+					mylogger.Error("DisableForIndexing", "addr", addr, "err", disableErr)
+				}
+				continue
+			}
+			return err
+		}
+
+		if updateErr := nftClassRepository.UpdateNFTClassesLatestEventBlockNumber(
+			ctx, []string{addr}, typeutil.Uint64(newBlockHeight),
+		); updateErr != nil {
+			mylogger.Error("UpdateNFTClassesLatestEventBlockNumber", "addr", addr, "err", updateErr)
+		}
+	}
+
+	return nil
+}
+
 func init() {
 	Tasks.Register(task.DefineTask(
 		TypeAcquireBookNFTEventsTaskPayload,

likenft-indexer/cmd/worker/task/acquire_book_nft_events_with_lifecycle.go

@@ -17,12 +17,25 @@ import (
 const TypeAcquireBookNFTEventsTaskPayloadWithLifecyclePayload = "acquire-book-nft-events-with-lifecycle"
 
 type AcquireBookNFTEventsTaskPayloadWithLifecyclePayload struct {
-	ContractAddress string
+	// Deprecated: use ContractAddresses instead. Kept for backward compatibility
+	// with in-flight tasks during rolling deploy.
+	ContractAddress   string   `json:"ContractAddress,omitempty"`
+	ContractAddresses []string `json:"ContractAddresses,omitempty"`
 }
 
-func NewTypeAcquireBookNFTEventsTaskPayloadWithLifecycle(contractAddress string) (*asynq.Task, error) {
+func (p *AcquireBookNFTEventsTaskPayloadWithLifecyclePayload) GetAddresses() []string {
+	if len(p.ContractAddresses) > 0 {
+		return p.ContractAddresses
+	}
+	if p.ContractAddress != "" {
+		return []string{p.ContractAddress}
+	}
+	return nil
+}
+
+func NewTypeAcquireBookNFTEventsTaskPayloadWithLifecycle(contractAddresses []string) (*asynq.Task, error) {
 	payload, err := json.Marshal(AcquireBookNFTEventsTaskPayloadWithLifecyclePayload{
-		ContractAddress: contractAddress,
+		ContractAddresses: contractAddresses,
 	})
 	if err != nil {
 		return nil, err
@@ -50,39 +63,72 @@ func handlerWithLifecycle(
 			return fmt.Errorf("json.Unmarshal: %v", err)
 		}
 
-		task, err := NewAcquireBookNFTEventsTask(
-			[]string{p.ContractAddress},
-		)
-		if err != nil {
-			return fmt.Errorf("NewAcquireBookNFTEventsTask: %v", err)
+		addresses := p.GetAddresses()
+		if len(addresses) == 0 {
+			return fmt.Errorf("no contract addresses in payload: %w", asynq.SkipRetry)
 		}
-		lifecycle, err := nftclassacquirebooknftevent.MakeNFTClassAcquireBookNFTEventLifecycleFromAddress(
-			ctx,
-			nftClassAcquireBookNFTEventsRepository,
-			p.ContractAddress,
-			nftclassacquirebooknftevent.MakeCalculateNextProcessingScoreFn(
-				config.TaskAcquireBookNFTNextProcessingBlockHeightWeight,
-				config.TaskAcquireBookNFTNextProcessingTimeFloor,
-				config.TaskAcquireBookNFTNextProcessingTimeCeiling,
-				config.TaskAcquireBookNFTNextProcessingTimeWeight,
-			),
-			nftclassacquirebooknftevent.MakeCalculateTimeoutScoreFn(
-				config.TaskAcquireBookNFTInProgressTimeoutSeconds,
-			),
-			nftclassacquirebooknftevent.MakeCalculateRetryScoreFn(
-				config.TaskAcquireBookNFTRetryInitialTimeoutSeconds,
-				config.TaskAcquireBookNFTRetryExponentialBackoffCoeff,
-				config.TaskAcquireBookNFTRetryMaxTimeoutSeconds,
-			),
+
+		calculateNextScoreFn := nftclassacquirebooknftevent.MakeCalculateNextProcessingScoreFn(
+			config.TaskAcquireBookNFTNextProcessingBlockHeightWeight,
+			config.TaskAcquireBookNFTNextProcessingTimeFloor,
+			config.TaskAcquireBookNFTNextProcessingTimeCeiling,
+			config.TaskAcquireBookNFTNextProcessingTimeWeight,
+		)
+		calculateTimeoutScoreFn := nftclassacquirebooknftevent.MakeCalculateTimeoutScoreFn(
+			config.TaskAcquireBookNFTInProgressTimeoutSeconds,
 		)
+		calculateRetryScoreFn := nftclassacquirebooknftevent.MakeCalculateRetryScoreFn(
+			config.TaskAcquireBookNFTRetryInitialTimeoutSeconds,
+			config.TaskAcquireBookNFTRetryExponentialBackoffCoeff,
+			config.TaskAcquireBookNFTRetryMaxTimeoutSeconds,
+		)
+
+		// Build lifecycle objects, tracking which addresses have valid lifecycles
+		var lifecycles []nftclassacquirebooknftevent.NFTClassAcquireBookNFTEventLifecycle
+		var validAddresses []string
+		for _, addr := range addresses {
+			lifecycle, err := nftclassacquirebooknftevent.MakeNFTClassAcquireBookNFTEventLifecycleFromAddress(
+				ctx,
+				nftClassAcquireBookNFTEventsRepository,
+				addr,
+				calculateNextScoreFn,
+				calculateTimeoutScoreFn,
+				calculateRetryScoreFn,
+			)
+			if err != nil {
+				mylogger.Error("MakeNFTClassAcquireBookNFTEventLifecycleFromAddress", "addr", addr, "err", err)
+				continue
+			}
+			lifecycles = append(lifecycles, lifecycle)
+			validAddresses = append(validAddresses, addr)
+		}
+
+		if len(lifecycles) == 0 {
+			return fmt.Errorf("no valid lifecycles for batch: %w", asynq.SkipRetry)
+		}
+
+		// Create innerTask with only addresses that have valid lifecycles
+		innerTask, err := NewAcquireBookNFTEventsTask(validAddresses)
 		if err != nil {
-			return fmt.Errorf("p.ToLifecycle: %v", err)
+			return fmt.Errorf("NewAcquireBookNFTEventsTask: %v", err)
 		}
 
-		if err := lifecycle.WithEnqueued(ctx, func(nftClass *ent.NFTClass) error {
-			return handler(ctx, task)
-		}); err != nil {
-			mylogger.Error("lifecycle.WithEnqueued", "err", err)
+		// Run the inner handler once, then transition all lifecycles.
+		// The first lifecycle executes the handler; the rest reuse its result.
+		// NOTE: This relies on WithEnqueued calling the callback synchronously.
+		var handlerErr error
+		handlerExecuted := false
+
+		for _, lifecycle := range lifecycles {
+			if err := lifecycle.WithEnqueued(ctx, func(nftClass *ent.NFTClass) error {
+				if !handlerExecuted {
+					handlerExecuted = true
+					handlerErr = handler(ctx, innerTask)
+				}
+				return handlerErr
+			}); err != nil {
+				mylogger.Error("lifecycle.WithEnqueued", "err", err)
+			}
 		}
 
 		return nil

likenft-indexer/cmd/worker/task/acquire_book_nft_events_with_lifecycle_test.go

@@ -0,0 +1,60 @@
+package task
+
+import "testing"
+
+func TestAcquireBookNFTEventsTaskPayloadWithLifecyclePayload_GetAddresses(t *testing.T) {
+	tests := []struct {
+		name     string
+		payload  AcquireBookNFTEventsTaskPayloadWithLifecyclePayload
+		expected []string
+	}{
+		{
+			name: "new field takes precedence",
+			payload: AcquireBookNFTEventsTaskPayloadWithLifecyclePayload{
+				ContractAddress:   "0xOLD",
+				ContractAddresses: []string{"0xA", "0xB"},
+			},
+			expected: []string{"0xA", "0xB"},
+		},
+		{
+			name: "falls back to old single address",
+			payload: AcquireBookNFTEventsTaskPayloadWithLifecyclePayload{
+				ContractAddress: "0xOLD",
+			},
+			expected: []string{"0xOLD"},
+		},
+		{
+			name:     "returns nil when both empty",
+			payload:  AcquireBookNFTEventsTaskPayloadWithLifecyclePayload{},
+			expected: nil,
+		},
+		{
+			name: "empty slice falls back to old address",
+			payload: AcquireBookNFTEventsTaskPayloadWithLifecyclePayload{
+				ContractAddress:   "0xOLD",
+				ContractAddresses: []string{},
+			},
+			expected: []string{"0xOLD"},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := tt.payload.GetAddresses()
+			if tt.expected == nil {
+				if got != nil {
+					t.Fatalf("got %v, want nil", got)
+				}
+				return
+			}
+			if len(got) != len(tt.expected) {
+				t.Fatalf("got %v, want %v", got, tt.expected)
+			}
+			for i := range got {
+				if got[i] != tt.expected[i] {
+					t.Fatalf("got[%d] = %q, want %q", i, got[i], tt.expected[i])
+				}
+			}
+		})
+	}
+}