Cache payer collateral with bounded canonical recovery

eth/payment.go

@@ -60,37 +60,39 @@ func (c PaymentChain) activeAt(ctx context.Context, address, recipient ethcommon
 	return bonding.IsActiveTranscoder(&bind.CallOpts{Context: ctx}, recipient)
 }
 
-// Receipt reports a submitted redemption without retrying or resubmitting it.
-func (c PaymentChain) Receipt(ctx context.Context, hash ethcommon.Hash) (confirmed, reverted bool, err error) {
+// FinalizedReceipt retains canonical inclusion evidence for durable collateral admission.
+func (c PaymentChain) FinalizedReceipt(ctx context.Context, hash ethcommon.Hash) (*types.Receipt, error) {
 	receipt, err := c.Contracts.RPC.TransactionReceipt(ctx, hash)
 	if errors.Is(err, ethereum.NotFound) {
-		return false, false, nil
+		return nil, nil
 	}
 	if err != nil {
-		return false, false, err
+		return nil, err
 	}
 	finalized, err := c.Contracts.RPC.header(ctx, "finalized")
 	if err != nil {
-		return false, false, err
+		return nil, err
 	}
 	if finalized.Number == nil {
-		return false, false, errors.New("invalid finalized header")
+		return nil, errors.New("invalid finalized header")
 	}
 	if (*big.Int)(finalized.Number).Cmp(receipt.BlockNumber) < 0 {
-		return false, false, nil
+		return nil, nil
 	}
 	canonical, err := c.Contracts.RPC.header(ctx, hexutil.EncodeBig(receipt.BlockNumber))
 	if err != nil {
-		return false, false, err
+		return nil, err
 	}
 	if canonical.Hash != receipt.BlockHash {
-		return false, false, nil
+		return nil, nil
 	}
-	return receipt.Status == types.ReceiptStatusSuccessful, receipt.Status == types.ReceiptStatusFailed, nil
+	return receipt, nil
 }
 
 type ChainSnapshot struct {
 	Block           *big.Int // Protocol L1 block, including on Arbitrum.
+	Number          *big.Int // Execution-chain height, distinct from the L1 clock.
+	BlockHash       ethcommon.Hash
 	Round           *big.Int // Last initialized round.
 	RoundHash       ethcommon.Hash
 	CurrentRound    *big.Int // Protocol round; may exceed the initialized Round.
@@ -158,7 +160,7 @@ func (c PaymentChain) snapshotAt(ctx context.Context, header *miniHeader, curren
 	if hash == ([32]byte{}) {
 		return ChainSnapshot{}, errors.New("invalid round block hash")
 	}
-	snapshot := ChainSnapshot{Block: block, Round: round, RoundHash: ethcommon.Hash(hash), blockReference: ref, addresses: addresses}
+	snapshot := ChainSnapshot{Block: block, Number: new(big.Int).Set((*big.Int)(header.Number)), BlockHash: header.Hash, Round: round, RoundHash: ethcommon.Hash(hash), blockReference: ref, addresses: addresses}
 	if currentRound {
 		snapshot.CurrentRound, err = rounds.CurrentRound(opts)
 		if err != nil {
@@ -183,10 +185,20 @@ type SenderInfo struct {
 }
 
 func (c PaymentChain) SenderInfo(ctx context.Context, sender, recipient ethcommon.Address) (SenderInfo, error) {
+	if c.Observer != nil {
+		if c.Contracts != c.Observer.chain.Contracts {
+			return SenderInfo{}, errors.New("payment observer belongs to another contract client")
+		}
+		return c.Observer.senderInfo(ctx, sender, recipient)
+	}
 	snapshot, err := c.Snapshot(ctx)
 	if err != nil {
 		return SenderInfo{}, err
 	}
+	return c.senderInfoAt(ctx, sender, recipient, snapshot)
+}
+
+func (c PaymentChain) senderInfoAt(ctx context.Context, sender, recipient ethcommon.Address, snapshot ChainSnapshot) (SenderInfo, error) {
 	address, err := c.Contracts.paymentAddress(ctx, snapshot, "ticketBroker")
 	if err != nil {
 		return SenderInfo{}, err
@@ -276,29 +288,32 @@ func (c PaymentChain) SubmitRedemption(ctx context.Context, nonces *NonceManager
 	return nil
 }
 
-// IsUsedTicket is ported from go-livepeer/eth/client_ticketbroker.go. The block
-// argument also supports the finalized observation needed for node settlement.
-func (c PaymentChain) IsUsedTicket(ctx context.Context, hash ethcommon.Hash, block string) (bool, error) {
+// TicketUse adapts go-livepeer/eth/client_ticketbroker.go's IsUsedTicket.
+// It returns the pinned finalized identity alongside ticket-use evidence.
+func (c PaymentChain) TicketUse(ctx context.Context, hash ethcommon.Hash, block string) (bool, ChainSnapshot, error) {
 	var ref any = block
+	var observation ChainSnapshot
 	if block == "finalized" {
 		header, err := c.Contracts.RPC.header(ctx, "finalized")
 		if err != nil {
-			return false, err
+			return false, ChainSnapshot{}, err
 		}
 		if header.Hash == (ethcommon.Hash{}) || header.Number == nil {
-			return false, errors.New("invalid finalized header")
+			return false, ChainSnapshot{}, errors.New("invalid finalized header")
 		}
 		ref = map[string]any{"blockHash": header.Hash, "requireCanonical": true}
+		observation = ChainSnapshot{Number: new(big.Int).Set((*big.Int)(header.Number)), BlockHash: header.Hash}
 	}
 	address, err := c.Contracts.ResolveAt(ctx, ref, "ticketBroker")
 	if err != nil {
-		return false, err
+		return false, ChainSnapshot{}, err
 	}
 	broker, err := contracts.NewTicketBrokerCaller(address, c.Contracts.callerAt(ref))
 	if err != nil {
-		return false, err
+		return false, ChainSnapshot{}, err
 	}
-	return broker.UsedTickets(&bind.CallOpts{Context: ctx}, hash)
+	used, err := broker.UsedTickets(&bind.CallOpts{Context: ctx}, hash)
+	return used, observation, err
 }
 
 // FinalizedNonce pins account state to the same canonical finalized reference

eth/payment_contracts_test.go

@@ -132,7 +132,7 @@ func TestObserverRegistryScopeAndFinalizedReads(t *testing.T) {
 	})
 	chain := o.chain
 	chain.Observer = o // Its retained registry is deliberately still the old one.
-	used, err := chain.IsUsedTicket(t.Context(), common.HexToHash("0x9999"), "finalized")
+	used, _, err := chain.TicketUse(t.Context(), common.HexToHash("0x9999"), "finalized")
 	require.NoError(t, err)
 	require.False(t, used, "finalized settlement resolves its own canonical registry")
 	require.Equal(t, 1, f.Stats().Calls["Controller.getContract"])

eth/payment_observer.go

@@ -35,6 +35,10 @@ type PaymentObserver struct {
 	l1Progress        time.Time
 	l1HighWater       *big.Int
 	recipient         common.Address
+	lifecycle         context.Context
+	payers            map[payerKey]*payerEntry
+	payerReads        sync.WaitGroup
+	activeReads       int
 }
 
 func NewPaymentObserver(chain PaymentChain, requireL1 bool, recipient common.Address) *PaymentObserver {
@@ -44,7 +48,17 @@ func NewPaymentObserver(chain PaymentChain, requireL1 bool, recipient common.Add
 
 // Run has one service-owned caller. Every RPC inherits the service lifetime.
 func (o *PaymentObserver) Run(ctx context.Context) {
-	defer func() { o.mu.Lock(); o.stopped = true; o.mu.Unlock() }()
+	o.mu.Lock()
+	lifecycle, cancelReads := context.WithCancel(ctx)
+	o.lifecycle = lifecycle
+	o.mu.Unlock()
+	defer func() {
+		cancelReads()
+		o.mu.Lock()
+		o.stopped = true
+		o.mu.Unlock()
+		o.payerReads.Wait()
+	}()
 	ticker := time.NewTicker(o.interval)
 	defer ticker.Stop()
 	for {
@@ -64,13 +78,20 @@ func (o *PaymentObserver) Run(ctx context.Context) {
 type ObserverStatus struct {
 	ObservedAt, ContactAt, HeadProgressAt, L1ProgressAt time.Time
 	Healthy, RecipientActive                            bool
+	PayerEntries, PayerReads, PayerFailures             int
 }
 
 func (o *PaymentObserver) Status() ObserverStatus {
 	o.mu.RLock()
 	defer o.mu.RUnlock()
 	healthy := o.validate(o.current) == nil
-	return ObserverStatus{o.current.observedAt, o.contact, o.progress, o.l1Progress, healthy, healthy && o.recipient != (common.Address{}) && o.current.recipientErr == nil && o.current.recipientActive}
+	status := ObserverStatus{ObservedAt: o.current.observedAt, ContactAt: o.contact, HeadProgressAt: o.progress, L1ProgressAt: o.l1Progress, Healthy: healthy, RecipientActive: healthy && o.recipient != (common.Address{}) && o.current.recipientErr == nil && o.current.recipientActive, PayerEntries: len(o.payers), PayerReads: o.activeReads}
+	for _, entry := range o.payers {
+		if !entry.valid && entry.failure != nil {
+			status.PayerFailures++
+		}
+	}
+	return status
 }
 
 func (o *PaymentObserver) Snapshot(ctx context.Context) (ChainSnapshot, error) {
@@ -82,9 +103,14 @@ func (o *PaymentObserver) Snapshot(ctx context.Context) (ChainSnapshot, error) {
 	if err := o.validate(o.current); err != nil {
 		return ChainSnapshot{}, err
 	}
-	s := o.current
-	s.Block, s.Round, s.CurrentRound = new(big.Int).Set(s.Block), new(big.Int).Set(s.Round), new(big.Int).Set(s.CurrentRound)
-	return s, nil
+	return o.current.clone(), nil
+}
+
+func (s ChainSnapshot) clone() ChainSnapshot {
+	for _, value := range []**big.Int{&s.Block, &s.Number, &s.Round, &s.CurrentRound} {
+		*value = new(big.Int).Set(*value)
+	}
+	return s
 }
 
 func (s ChainSnapshot) ObservedAt() time.Time {
@@ -106,7 +132,7 @@ func (s ChainSnapshot) Validate() error {
 }
 
 func (o *PaymentObserver) validate(s ChainSnapshot) error {
-	if o.stopped || o.invalid || s.generation != o.generation || o.expired(s, o.l1Progress) {
+	if o.stopped || o.invalid || (o.lifecycle != nil && o.lifecycle.Err() != nil) || s.generation != o.generation || o.expired(s, o.l1Progress) {
 		return ErrObservationUnavailable
 	}
 	return nil
@@ -117,8 +143,7 @@ func (o *PaymentObserver) expired(s ChainSnapshot, l1Progress time.Time) bool {
 	return s.observedAt.IsZero() || now.Sub(s.observedAt) >= o.maxAge || now.Sub(s.blockTime) >= o.maxAge || now.Sub(l1Progress) >= o.maxAge
 }
 
-// Only Run calls refresh. Previous/current identity is the extension point for
-// complete hash-pinned logs or reset; no partial delta authorizes publication.
+// Only Run calls refresh.
 func (o *PaymentObserver) refresh(ctx context.Context) error {
 	at := o.now()
 	head, err := o.chain.Contracts.RPC.header(ctx, "latest")
@@ -128,7 +153,7 @@ func (o *PaymentObserver) refresh(ctx context.Context) error {
 	o.mu.Lock()
 	o.contact = o.now()
 	prior := o.header
-	gap := false
+	gap, continuous := false, false
 	if head.Hash == (common.Hash{}) || head.Number == nil {
 		o.mu.Unlock()
 		return ErrObservationUnavailable
@@ -142,6 +167,7 @@ func (o *PaymentObserver) refresh(ctx context.Context) error {
 			return ErrObservationUnavailable
 		}
 		gap = head.Hash != prior.Hash && distance.Cmp(big.NewInt(1)) > 0
+		continuous = distance.Cmp(big.NewInt(1)) == 0 && head.ParentHash == prior.Hash
 		if head.Hash != prior.Hash && !gap && (distance.Sign() == 0 || head.ParentHash != prior.Hash) {
 			o.invalid = true
 			o.generation++
@@ -171,7 +197,8 @@ func (o *PaymentObserver) refresh(ctx context.Context) error {
 		// A polling gap does not itself orphan the retained observation. Prove
 		// its anchor before keeping it available during the pinned backfill.
 		anchor, err := o.chain.Contracts.RPC.header(ctx, hexutil.EncodeBig((*big.Int)(prior.Number)))
-		if err != nil || anchor.Hash != prior.Hash || anchor.Number == nil || (*big.Int)(anchor.Number).Cmp((*big.Int)(prior.Number)) != 0 {
+		continuous = err == nil && anchor.Hash == prior.Hash && anchor.Number != nil && (*big.Int)(anchor.Number).Cmp((*big.Int)(prior.Number)) == 0
+		if !continuous {
 			o.mu.Lock()
 			o.invalid = true
 			o.generation++
@@ -193,6 +220,19 @@ func (o *PaymentObserver) refresh(ctx context.Context) error {
 	if err := ctx.Err(); err != nil {
 		return err
 	}
+	changed, reset := o.payerChanges(ctx, prior, head, s, continuous)
+	if gap && reset {
+		o.invalidate()
+	}
+	// Hash-pinned logs can outlive their canonical block, even on a reset.
+	canonical, err := o.chain.Contracts.RPC.header(ctx, hexutil.EncodeBig((*big.Int)(head.Number)))
+	if err != nil || canonical.Hash != head.Hash {
+		o.invalidate()
+		return ErrObservationUnavailable
+	}
+	if err := ctx.Err(); err != nil {
+		return err
+	}
 	o.mu.Lock()
 	defer o.mu.Unlock()
 	progress, l1Progress, highWater := o.progress, o.l1Progress, o.l1HighWater
@@ -206,6 +246,12 @@ func (o *PaymentObserver) refresh(ctx context.Context) error {
 	if o.stopped || o.expired(s, l1Progress) {
 		return ErrObservationUnavailable
 	}
+	for key, entry := range o.payers {
+		if _, affected := changed[key.payer]; reset || affected {
+			entry.version++
+			entry.valid, entry.failure = false, nil
+		}
+	}
 	o.current, o.header, o.invalid = s, head, false
 	o.progress, o.l1Progress, o.l1HighWater = progress, l1Progress, highWater
 	o.requireL1 = o.requireL1 || head.L1BlockNumber != nil

eth/payment_observer_test.go

@@ -26,6 +26,7 @@ func newTestObserver(t *testing.T) (*test.ChainRPC, *PaymentObserver, *time.Time
 	c, err := NewContracts(rpc, f.Controller)
 	require.NoError(t, err)
 	o := NewPaymentObserver(PaymentChain{Contracts: c}, true, common.Address{})
+	o.lifecycle = t.Context()
 	now := time.Now().Truncate(time.Second)
 	o.now = func() time.Time { return now }
 	return f, o, &now
@@ -36,6 +37,7 @@ func advanceObserver(f *test.ChainRPC, now time.Time) {
 		s.Parent, s.Number, s.Timestamp = s.Hash, s.Number+1, uint64(now.Unix())
 		s.Hash = common.BigToHash(new(big.Int).SetUint64(s.Number))
 		s.L1Block++
+		s.Logs, s.LogsBloom = nil, nil
 	})
 }
 
@@ -202,23 +204,27 @@ func TestObserverL1ProgressDeadline(t *testing.T) {
 }
 
 func TestObserverExpiredCandidateIsNotPublished(t *testing.T) {
-	f, o, now := newTestObserver(t)
-	require.NoError(t, o.refresh(t.Context()))
-	old := o.Status()
-	advanceObserver(f, *now)
-	f.Update(func(s *test.ChainState) {
-		s.Before = func(_ context.Context, name string) {
-			if name == "RoundsManager.currentRound" {
-				*now = now.Add(30 * time.Second)
-			}
-		}
-	})
-	require.ErrorIs(t, o.refresh(t.Context()), ErrObservationUnavailable)
-	status := o.Status()
-	require.Equal(t, old.ObservedAt, status.ObservedAt)
-	require.Equal(t, old.HeadProgressAt, status.HeadProgressAt)
-	require.Equal(t, old.L1ProgressAt, status.L1ProgressAt)
-	require.Equal(t, int64(50), o.current.Block.Int64())
+	for _, phase := range []string{"RoundsManager.currentRound", "eth_getLogs"} {
+		t.Run(phase, func(t *testing.T) {
+			f, o, now := newTestObserver(t)
+			require.NoError(t, o.refresh(t.Context()))
+			old := o.Status()
+			advanceObserver(f, *now)
+			f.Update(func(s *test.ChainState) {
+				s.Before = func(_ context.Context, name string) {
+					if name == phase {
+						*now = now.Add(30 * time.Second)
+					}
+				}
+			})
+			require.ErrorIs(t, o.refresh(t.Context()), ErrObservationUnavailable)
+			status := o.Status()
+			require.Equal(t, old.ObservedAt, status.ObservedAt)
+			require.Equal(t, old.HeadProgressAt, status.HeadProgressAt)
+			require.Equal(t, old.L1ProgressAt, status.L1ProgressAt)
+			require.Equal(t, int64(50), o.current.Block.Int64())
+		})
+	}
 }
 
 func TestObserverPinnedBackfillAndReorg(t *testing.T) {

eth/payment_test.go

@@ -168,16 +168,20 @@ func TestRedemptionReceiptRequiresCanonicalFinality(t *testing.T) {
 			require.NoError(t, err)
 			defer rpc.Close()
 			contracts := &Contracts{RPC: rpc}
-			confirmed, reverted, err := (PaymentChain{Contracts: contracts}).Receipt(t.Context(), ethcommon.HexToHash("0x1234"))
+			receipt, err := (PaymentChain{Contracts: contracts}).FinalizedReceipt(t.Context(), ethcommon.HexToHash("0x1234"))
 			if strings.HasPrefix(scenario, "invalid") {
 				require.ErrorContains(t, err, "invalid transaction receipt")
 				_, err = contracts.WaitReceipt(t.Context(), ethcommon.HexToHash("0x1234"))
 				require.ErrorContains(t, err, "invalid transaction receipt")
 			} else {
 				require.NoError(t, err)
 			}
-			require.Equal(t, scenario == "confirmed", confirmed)
-			require.Equal(t, scenario == "reverted", reverted)
+			require.Equal(t, scenario == "confirmed", receipt != nil && receipt.Status == types.ReceiptStatusSuccessful)
+			require.Equal(t, scenario == "reverted", receipt != nil && receipt.Status == types.ReceiptStatusFailed)
+			if receipt != nil {
+				require.Equal(t, big.NewInt(16), receipt.BlockNumber)
+				require.Equal(t, hash, receipt.BlockHash.Hex())
+			}
 		})
 	}
 }

eth/rpc.go

@@ -226,6 +226,7 @@ type miniHeader struct {
 	Hash          ethcommon.Hash  `json:"hash"`
 	ParentHash    ethcommon.Hash  `json:"parentHash"`
 	Timestamp     *hexutil.Uint64 `json:"timestamp"`
+	LogsBloom     *types.Bloom    `json:"logsBloom"`
 }
 
 func (r *RPC) header(ctx context.Context, block any) (*miniHeader, error) {