feat: exclude offer-locked holdings from indexed balances (#340)

* feat(indexer): exclude offer-locked holdings from balances

Offer-based transfers (USDCx/AllocationFactory) lock the sender's holding when
an offer is created; the locked amount is escrowed, not spendable. The indexer
derived balances purely from holding create/archive, so the sender's balance
only dropped at accept — the offered amount still showed as available while the
offer was pending.

Capture the holding's `lock` field in the decoder and skip locked holdings in
the processor (neither stored nor counted). Because a locked holding is never
stored, its later archive is a no-op, and the matching unlocked holding — created
for the receiver on accept or returned to the sender on claim-back — drives the
balance change. Net effect: the sender's balance is reduced when the offer locks
the funds, credited to the receiver on accept, and restored to the sender on
claim-back. Direct CIP-56 transfers are unaffected (their holdings are unlocked).

No migration: locked holdings are simply not persisted.

Implements #338.

* test(e2e): assert offer-time balance deduction in P1↔P1 USDCx transfer

Extend TestUSDCx_InternalTransfer_P1HolderToP1Holder: after User1 offers 10 of
its 15 USDCx to User2 (and before User2 accepts), assert User1's spendable
balance has already dropped to 5 — confirming the indexer excludes the locked
(escrowed) holding from balances at offer time, not only on accept.

Adds WaitForAPIBalanceExact (the existing >= helper can't detect a decrease).

Author: sadiq1971

pkg/indexer/engine/decoder.go

@@ -221,6 +221,9 @@ func NewHoldingDecoder(
 			change.InstrumentAdmin = ev.PartyField("registrar")
 			change.InstrumentID = ev.NestedTextField("instrument", "id")
 			change.Amount = ev.NumericField("amount")
+			// `lock` is Optional Lock: Some => the holding is escrowed by an
+			// outstanding offer (not spendable), None => freely spendable.
+			change.Locked = !ev.IsNone("lock")
 			if change.Owner == "" || change.InstrumentID == "" {
 				logger.Warn("Holding CREATED decoded with empty owner or instrument — field-name mismatch?",
 					zap.String("contract_id", ev.ContractID),

pkg/indexer/engine/decoder_test.go

@@ -75,6 +75,29 @@ func decodeAll(decode func(*streaming.LedgerTransaction, *streaming.LedgerEvent)
 // Decoder tests
 // ---------------------------------------------------------------------------
 
+func makeHoldingEvent(contractID string, lock streaming.FieldValue) *streaming.LedgerEvent {
+	return streaming.NewLedgerEvent(contractID, "pkg-id", holdingModule, holdingEntity, true,
+		map[string]streaming.FieldValue{
+			"owner":      streaming.MakePartyField("alice::1220"),
+			"registrar":  streaming.MakePartyField("admin::1220"),
+			"instrument": streaming.MakeRecordField(map[string]streaming.FieldValue{"id": streaming.MakeTextField("USDCx")}),
+			"amount":     streaming.MakeNumericField("10"),
+			"lock":       lock,
+		})
+}
+
+func TestHoldingDecoder_LockField(t *testing.T) {
+	dec := NewHoldingDecoder("pkg-id", zap.NewNop())
+
+	unlocked, ok := dec(makeTx(1), makeHoldingEvent("h-unlocked", streaming.MakeNoneField()))
+	require.True(t, ok)
+	assert.False(t, unlocked.Locked, "None lock => spendable")
+
+	locked, ok := dec(makeTx(2), makeHoldingEvent("h-locked", streaming.MakeSomeRecordField(map[string]streaming.FieldValue{})))
+	require.True(t, ok)
+	assert.True(t, locked.Locked, "Some lock => escrowed")
+}
+
 func TestDecoder_FilterModeAll_Mint(t *testing.T) {
 	decode := NewTokenTransferDecoder(indexer.FilterModeAll, nil, zap.NewNop())
 

pkg/indexer/engine/processor.go

@@ -369,6 +369,15 @@ func (p *Processor) processHoldingChange(ctx context.Context, tx Store, h *index
 		// Malformed CREATED event already logged by decoder; skip silently.
 		return nil
 	}
+	if h.Locked {
+		// Escrowed by an outstanding transfer offer — not spendable, so excluded from
+		// balances. We also don't store it: its archive (on accept or claim-back) then
+		// finds no row and is skipped, and the matching unlocked holding created for
+		// the receiver (accept) or returned to the sender (claim-back) drives the
+		// balance change. Net effect: the sender's balance drops when the offer locks
+		// the funds and is restored only if the offer is claimed back.
+		return nil
+	}
 	if err := tx.InsertHolding(ctx, h); err != nil {
 		return fmt.Errorf("insert holding %s: %w", h.ContractID, err)
 	}

pkg/indexer/engine/processor_test.go

@@ -106,6 +106,18 @@ func makeOfferBatch(offset int64, offers ...*indexer.Transfer) *streaming.Batch[
 	}
 }
 
+func makeHoldingBatch(offset int64, holdings ...*indexer.HoldingChange) *streaming.Batch[any] {
+	items := make([]any, len(holdings))
+	for i, h := range holdings {
+		items[i] = h
+	}
+	return &streaming.Batch[any]{
+		Offset:   offset,
+		UpdateID: "update-" + string(rune('0'+offset)),
+		Items:    items,
+	}
+}
+
 func pendingOffer() *indexer.Transfer {
 	return &indexer.Transfer{
 		ContractID:      "offer-contract-1",
@@ -419,6 +431,60 @@ func TestProcessor_Run_OfferArchived_CompletesTransfer(t *testing.T) {
 	require.NoError(t, engine.NewProcessor(fetcher, store, engine.NewNopMetrics(), zap.NewNop()).Run(context.Background()))
 }
 
+func TestProcessor_Run_LockedHoldingCreated_SkippedFromBalance(t *testing.T) {
+	store := mocks.NewStore(t)
+	fetcher := mocks.NewEventFetcher(t)
+	// A locked holding is escrowed by an outstanding offer; it must not be stored or
+	// counted toward balances (so the sender's balance reflects the offered amount as
+	// already deducted from offer creation).
+	locked := &indexer.HoldingChange{
+		ContractID:      "holding-locked-1",
+		Owner:           testSender,
+		InstrumentAdmin: testInstrumentAdmin,
+		InstrumentID:    testInstrumentID,
+		Amount:          testAmount,
+		LedgerOffset:    12,
+		Locked:          true,
+	}
+
+	store.EXPECT().LatestOffset(mock.Anything).Return(int64(0), nil)
+	fetcher.EXPECT().Start(mock.Anything, int64(0))
+	fetcher.EXPECT().Events().Return(feedCh(makeHoldingBatch(12, locked)))
+
+	setupRunInTx(store)
+	// No InsertHolding / ApplyBalanceDelta expected: the strict mock fails if either
+	// is called. Only the offset advances.
+	store.EXPECT().SaveOffset(mock.Anything, int64(12)).Return(nil)
+
+	require.NoError(t, engine.NewProcessor(fetcher, store, engine.NewNopMetrics(), zap.NewNop()).Run(context.Background()))
+}
+
+func TestProcessor_Run_UnlockedHoldingCreated_Tracked(t *testing.T) {
+	store := mocks.NewStore(t)
+	fetcher := mocks.NewEventFetcher(t)
+	// An unlocked (spendable) holding is stored and credited to the owner's balance.
+	h := &indexer.HoldingChange{
+		ContractID:      "holding-1",
+		Owner:           testRecipient,
+		InstrumentAdmin: testInstrumentAdmin,
+		InstrumentID:    testInstrumentID,
+		Amount:          testAmount,
+		LedgerOffset:    13,
+		Locked:          false,
+	}
+
+	store.EXPECT().LatestOffset(mock.Anything).Return(int64(0), nil)
+	fetcher.EXPECT().Start(mock.Anything, int64(0))
+	fetcher.EXPECT().Events().Return(feedCh(makeHoldingBatch(13, h)))
+
+	setupRunInTx(store)
+	store.EXPECT().InsertHolding(mock.Anything, h).Return(nil)
+	store.EXPECT().ApplyBalanceDelta(mock.Anything, testRecipient, testInstrumentAdmin, testInstrumentID, testAmount).Return(nil)
+	store.EXPECT().SaveOffset(mock.Anything, int64(13)).Return(nil)
+
+	require.NoError(t, engine.NewProcessor(fetcher, store, engine.NewNopMetrics(), zap.NewNop()).Run(context.Background()))
+}
+
 func TestProcessor_Run_MixedBatch_ParsedEventAndOffer(t *testing.T) {
 	store := mocks.NewStore(t)
 	fetcher := mocks.NewEventFetcher(t)

pkg/indexer/types.go

@@ -141,6 +141,12 @@ type HoldingChange struct {
 	InstrumentAdmin string
 	InstrumentID    string
 	Amount          string
+	// Locked is true when the holding carries a lock (DAML `lock` is Some) — i.e. it
+	// is escrowed by an outstanding transfer offer rather than spendable. Locked
+	// holdings are excluded from indexed balances so a sender's balance reflects the
+	// offered amount as deducted from offer creation until accept or claim-back.
+	// Only meaningful for CREATED events.
+	Locked bool
 }
 
 // InstrumentKey is the Canton equivalent of an ERC-20 contract address.

tests/e2e/devstack/dsl/dsl.go

@@ -232,6 +232,46 @@ func (d *DSL) WaitForAPIBalance(ctx context.Context, t *testing.T, tok *stack.To
 		minTokens, tok.Symbol, minWei, ownerAddr.Hex())
 }
 
+// WaitForAPIBalanceExact polls the api-server until ownerAddr's balance of tok equals
+// exactly tokens (scaled by the token's decimals), or the timeout is reached. Unlike
+// WaitForAPIBalance (>=), this detects a decrease — e.g. funds becoming unspendable
+// when an offer locks them — so it converges on the exact expected value.
+func (d *DSL) WaitForAPIBalanceExact(ctx context.Context, t *testing.T, tok *stack.Token, ownerAddr common.Address, tokens string) {
+	t.Helper()
+	exp := new(big.Int).Exp(big.NewInt(decimalBase), big.NewInt(int64(tok.Decimals)), nil)
+	wantF, ok := new(big.Float).SetString(tokens)
+	if !ok {
+		t.Fatalf("WaitForAPIBalanceExact: invalid amount %q", tokens)
+	}
+	wantF.Mul(wantF, new(big.Float).SetInt(exp))
+	wantWei, _ := wantF.Int(nil)
+
+	deadline := time.Now().Add(waitForAPIBalanceTimeout)
+	ticker := time.NewTicker(pollInterval)
+	defer ticker.Stop()
+	var lastBal *big.Int
+	for time.Now().Before(deadline) {
+		bal, err := d.apiServer.ERC20Balance(ctx, tok.Address, ownerAddr)
+		if err == nil {
+			lastBal = bal
+			if bal.Cmp(wantWei) == 0 {
+				return
+			}
+		}
+		select {
+		case <-ctx.Done():
+			t.Fatal("context canceled waiting for exact API balance")
+		case <-ticker.C:
+		}
+	}
+	last := "<none>"
+	if lastBal != nil {
+		last = lastBal.String()
+	}
+	t.Fatalf("WaitForAPIBalanceExact: timed out waiting for %s %s balance == %s (owner %s): last seen %s",
+		tokens, tok.Symbol, wantWei, ownerAddr.Hex(), last)
+}
+
 // ERC20Balance returns the on-chain ERC-20 balance of account for tokenAddr.
 func (d *DSL) ERC20Balance(ctx context.Context, t *testing.T, tokenAddr common.Address, account stack.Account) *big.Int {
 	t.Helper()

tests/e2e/tests/api/cross_transfer_test.go

@@ -187,6 +187,12 @@ func TestUSDCx_InternalTransfer_P1HolderToP1Holder(t *testing.T) {
 		t.Fatalf("expected status 'completed', got %q", execResp.Status)
 	}
 
+	// Offering locks User1's 10 USDCx: the indexer excludes the escrowed (locked)
+	// holding from balances, so User1's spendable balance drops 15 → 5 immediately at
+	// offer time — before User2 accepts. (Without offer-locked accounting it would
+	// still read 15 here and only drop on accept.)
+	sys.DSL.WaitForAPIBalanceExact(ctx, t, &sys.Tokens.USDCx, sys.Accounts.User1.Address, "5")
+
 	// User2 accepts the inbound offer via the api-server incoming accept flow.
 	cid2 := sys.DSL.WaitForIncomingTransferOffer(ctx, t, sys.Accounts.User2)
 	prepAccept2, err := sys.APIServer.PrepareAcceptTransfer(ctx, &sys.Accounts.User2, cid2,