diff --git a/eth/payment.go b/eth/payment.go index 07273ba..b12fc8f 100644 --- a/eth/payment.go +++ b/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 diff --git a/eth/payment_contracts_test.go b/eth/payment_contracts_test.go index 98126ef..607ecba 100644 --- a/eth/payment_contracts_test.go +++ b/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"]) diff --git a/eth/payment_observer.go b/eth/payment_observer.go index 2baf7b6..00036ba 100644 --- a/eth/payment_observer.go +++ b/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 diff --git a/eth/payment_observer_test.go b/eth/payment_observer_test.go index f133b08..1b179d0 100644 --- a/eth/payment_observer_test.go +++ b/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) { diff --git a/eth/payment_test.go b/eth/payment_test.go index 4ccaaed..8846872 100644 --- a/eth/payment_test.go +++ b/eth/payment_test.go @@ -168,7 +168,7 @@ 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")) @@ -176,8 +176,12 @@ func TestRedemptionReceiptRequiresCanonicalFinality(t *testing.T) { } 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()) + } }) } } diff --git a/eth/rpc.go b/eth/rpc.go index d023838..6c79499 100644 --- a/eth/rpc.go +++ b/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) { diff --git a/eth/senderwatcher.go b/eth/senderwatcher.go new file mode 100644 index 0000000..bc329aa --- /dev/null +++ b/eth/senderwatcher.go @@ -0,0 +1,240 @@ +// Sender-key event invalidation and pinned recovery adapt livepeer/go-livepeer +// eth/watchers/senderwatcher.go (341b39a0c, 57e288706), by Nico Vergauwen. +// Immutable results, idle eviction and shared cold reads adapt Mike Zupper's +// BlueClawNetwork/livepeer-node eth/chainview.go (898977754, 5f20cab0f). +// Copyright (c) 2017 Livepeer; MIT notice in LICENSE. +package eth + +import ( + "context" + "errors" + "math/big" + "time" + + "github.com/ethereum/go-ethereum" + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/common/hexutil" +) + +const payerEntries = 1024 +const payerReadLimit = 4 // No distinct-key queue; cancelled jobs retain their slot until exit. +const payerIdle = 10 * time.Minute + +type payerKey struct{ payer, recipient common.Address } +type payerEntry struct { + info SenderInfo + valid bool + used time.Time + version uint64 + failure error + failedAt common.Hash + failedGeneration uint64 + job *payerRead +} +type payerRead struct { + done chan struct{} + cancel context.CancelFunc + waiters int + info SenderInfo + err error +} + +func (i SenderInfo) clone() SenderInfo { + i.Snapshot = i.Snapshot.clone() + i.Deposit, i.Reserve, i.WithdrawRound = new(big.Int).Set(i.Deposit), new(big.Int).Set(i.Reserve), new(big.Int).Set(i.WithdrawRound) + return i +} + +func (o *PaymentObserver) senderInfo(ctx context.Context, payer, recipient common.Address) (SenderInfo, error) { + if err := ctx.Err(); err != nil { + return SenderInfo{}, err + } + o.mu.Lock() + if err := o.validate(o.current); err != nil { + o.mu.Unlock() + return SenderInfo{}, err + } + key, now := payerKey{payer, recipient}, o.now() + entry := o.payers[key] + if entry == nil { + if o.activeReads >= payerReadLimit { + o.mu.Unlock() + return SenderInfo{}, ErrObservationUnavailable + } + var oldest payerKey + var last time.Time + for k, e := range o.payers { + if e.job == nil && now.Sub(e.used) >= payerIdle { + delete(o.payers, k) + continue + } + if e.job == nil && (last.IsZero() || e.used.Before(last)) { + oldest, last = k, e.used + } + } + if len(o.payers) >= payerEntries { + delete(o.payers, oldest) + } + if o.payers == nil { + o.payers = make(map[payerKey]*payerEntry) + } + entry = &payerEntry{} + o.payers[key] = entry + } + entry.used = now + if entry.valid { + info := entry.info + info.Snapshot = o.current + o.mu.Unlock() + return info.clone(), nil + } + if entry.failure != nil && entry.failedAt == o.current.BlockHash && entry.failedGeneration == o.current.generation { + err := entry.failure + o.mu.Unlock() + return SenderInfo{}, err + } + job := entry.job + if job == nil { + if o.activeReads >= payerReadLimit { + o.mu.Unlock() + return SenderInfo{}, ErrObservationUnavailable + } + read, cancel := context.WithTimeout(o.lifecycle, 15*time.Second) + job = &payerRead{done: make(chan struct{}), cancel: cancel} + entry.job = job + o.activeReads++ + o.payerReads.Add(1) + go o.readPayer(read, key, entry, job, entry.version, o.current) + } + job.waiters++ + o.mu.Unlock() + select { + case <-ctx.Done(): + case <-job.done: + } + o.mu.Lock() + job.waiters-- + if job.waiters == 0 { + job.cancel() + } + o.mu.Unlock() + if err := ctx.Err(); err != nil { + return SenderInfo{}, err + } + if job.err != nil { + if errors.Is(job.err, context.Canceled) { + return SenderInfo{}, ErrObservationUnavailable + } + return SenderInfo{}, job.err + } + if err := job.info.Snapshot.Validate(); err != nil { + return SenderInfo{}, err + } + return job.info.clone(), nil +} + +func (o *PaymentObserver) readPayer(ctx context.Context, key payerKey, entry *payerEntry, job *payerRead, version uint64, snapshot ChainSnapshot) { + defer o.payerReads.Done() + defer job.cancel() + info, err := o.chain.senderInfoAt(ctx, key.payer, key.recipient, snapshot) + o.mu.Lock() + defer o.mu.Unlock() + if err == nil { + err = ctx.Err() + } + if err == nil { + err = o.validate(snapshot) + } + if err == nil && entry.version != version { + err = ErrObservationUnavailable + } + if err == nil { + // Complete intervening logs left this entry unchanged; advance metadata too. + info.Snapshot = o.current + entry.info, entry.valid = info, true + entry.failure = nil + } else if ctx.Err() == nil && entry.version == version && !errors.Is(err, ErrObservationUnavailable) { + entry.failure, entry.failedAt, entry.failedGeneration = err, snapshot.BlockHash, snapshot.generation + } + job.info, job.err = info, err + entry.job = nil + o.activeReads-- + close(job.done) +} + +func (o *PaymentObserver) invalidate() { + o.mu.Lock() + o.invalid = true + o.generation++ + o.mu.Unlock() +} + +// Link each intervening block to the retained anchor and read its logs by hash. +// Incomplete history resets entries; refresh confirms the final canonical head. +func (o *PaymentObserver) payerChanges(ctx context.Context, prior, head *miniHeader, snapshot ChainSnapshot, continuous bool) (map[common.Address]bool, bool) { + o.mu.RLock() + reset := !continuous || (prior != nil && new(big.Int).Sub((*big.Int)(head.Number), (*big.Int)(prior.Number)).Cmp(big.NewInt(128)) > 0) || + o.current.CurrentRound.Cmp(snapshot.CurrentRound) != 0 || o.current.addresses != snapshot.addresses + o.mu.RUnlock() + if reset { + return nil, true + } + addresses := []common.Address{snapshot.addresses.broker, snapshot.addresses.bonding, o.chain.Contracts.Controller} + changed := make(map[common.Address]bool) + broker, bonding := o.chain.Contracts.abis["ticketBroker"], o.chain.Contracts.abis["bondingManager"] + block := head + for remaining := new(big.Int).Sub((*big.Int)(head.Number), (*big.Int)(prior.Number)).Int64(); remaining > 0; remaining-- { + if remaining == 1 && block.ParentHash != prior.Hash { + return nil, true + } + if bloom := block.LogsBloom; bloom == nil || + bloom.Test(addresses[0].Bytes()) || + bloom.Test(addresses[1].Bytes()) || + bloom.Test(addresses[2].Bytes()) { + logs, err := o.chain.Contracts.RPC.ethereum.FilterLogs(ctx, ethereum.FilterQuery{BlockHash: &block.Hash, Addresses: addresses}) + if err != nil { + return nil, true + } + for _, log := range logs { + if log.Removed || log.BlockHash != block.Hash || log.BlockNumber != (*big.Int)(block.Number).Uint64() || len(log.Topics) == 0 { + return nil, true + } + if log.Address == addresses[2] { + // Proxy targets can change while all three public addresses stay fixed. + if log.Topics[0] == o.chain.Contracts.abis["controller"].Events["SetContractInfo"].ID { + return nil, true + } + continue + } + if log.Address == addresses[1] { + for _, name := range []string{"TranscoderActivated", "TranscoderDeactivated", "ParameterUpdate", "SetController"} { + if log.Topics[0] == bonding.Events[name].ID { + return nil, true + } + } + continue + } + for _, name := range []string{"DepositFunded", "ReserveFunded", "ReserveClaimed", "Withdrawal", "WinningTicketTransfer", "Unlock", "UnlockCancelled"} { + if log.Topics[0] == broker.Events[name].ID { + if len(log.Topics) < 2 { + return nil, true + } + changed[common.BytesToAddress(log.Topics[1].Bytes())] = true + } + } + if log.Topics[0] == broker.Events["SetController"].ID { + return nil, true + } + } + } + if remaining > 1 { + number := new(big.Int).Sub((*big.Int)(block.Number), big.NewInt(1)) + parent, err := o.chain.Contracts.RPC.header(ctx, hexutil.EncodeBig(number)) + if err != nil || parent.Number == nil || (*big.Int)(parent.Number).Cmp(number) != 0 || parent.Hash != block.ParentHash { + return nil, true + } + block = parent + } + } + return changed, false +} diff --git a/eth/senderwatcher_test.go b/eth/senderwatcher_test.go new file mode 100644 index 0000000..bfd7872 --- /dev/null +++ b/eth/senderwatcher_test.go @@ -0,0 +1,750 @@ +package eth + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "math/big" + "net/http" + "net/url" + "sync/atomic" + "testing" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/common/hexutil" + "github.com/ethereum/go-ethereum/core/types" + "github.com/livepeer/node/internal/test" + "github.com/stretchr/testify/require" +) + +// Cases adapt Nico Vergauwen's go-livepeer eth/watchers/senderwatcher_test.go +// (341b39a0c), and Mike Zupper's PR #1 eth/chainview_test.go (bbe126e). +// Copyright (c) 2017 Livepeer; MIT notice in LICENSE. +func warmPayer(t *testing.T, o *PaymentObserver, payer, recipient common.Address) SenderInfo { + t.Helper() + require.NoError(t, o.refresh(t.Context())) + funds, err := o.senderInfo(t.Context(), payer, recipient) + require.NoError(t, err) + return funds +} + +func payerHistoryBloom(s *test.ChainState, mode string) { + s.LogsBloom = nil + switch mode { + case "empty": + s.LogsBloom = types.Bloom{} + case "null": + s.LogsBloom = json.RawMessage("null") + case "malformed": + s.LogsBloom = "0x01" + case "positive", "unrelated": + address := s.Contracts["TicketBroker"] + if mode == "unrelated" { + address = common.HexToAddress("0x9999") + } + s.Logs = []types.Log{{Address: address, BlockHash: s.Hash, BlockNumber: s.Number, Topics: []common.Hash{common.HexToHash("0x1")}}} + s.LogsBloom = payerLogsBloom(s.Logs) + } +} + +func payerLogsBloom(logs []types.Log) types.Bloom { + receipt := new(types.Receipt) + for n := range logs { + receipt.Logs = append(receipt.Logs, &logs[n]) + } + return types.CreateBloom(receipt) +} + +func TestPayerCacheWarmCopiesAndEntryFailure(t *testing.T) { + f, o, now := newTestObserver(t) + payer, recipient := common.HexToAddress("0x11"), common.HexToAddress("0x22") + o.recipient = recipient + f.Update(func(s *test.ChainState) { s.Claimable = 7; s.Reserve = 50; s.Recipient = recipient }) + info := warmPayer(t, o, payer, recipient) + c := PaymentChain{Contracts: o.chain.Contracts, Observer: o} + require.Equal(t, int64(7), info.Reserve.Int64()) + info.Deposit.SetInt64(0) + info.Snapshot.Number.SetInt64(0) + before := f.Stats() + for range 20 { + info, err := c.SenderInfo(t.Context(), payer, recipient) + require.NoError(t, err) + require.Positive(t, info.Deposit.Sign()) + require.Positive(t, info.Snapshot.Number.Sign()) + active, err := c.IsActiveAt(t.Context(), recipient, info.Snapshot) + require.NoError(t, err) + require.True(t, active, "payer and activity share one complete observation") + } + require.Equal(t, before, f.Stats()) + foreign := PaymentChain{Contracts: &Contracts{}, Observer: o} + _, err := foreign.SenderInfo(t.Context(), payer, recipient) + require.ErrorContains(t, err, "another contract client") + require.Equal(t, before, f.Stats()) + total, err := c.SenderInfo(t.Context(), payer, common.Address{}) + require.NoError(t, err) + require.Equal(t, int64(50), total.Reserve.Int64()) + bad := common.HexToAddress("0xbad") + f.Update(func(s *test.ChainState) { s.FailPayer = bad }) + _, err = c.SenderInfo(t.Context(), bad, recipient) + require.Error(t, err) + before = f.Stats() + _, err = c.SenderInfo(t.Context(), bad, recipient) + require.Error(t, err) + _, err = c.SenderInfo(t.Context(), payer, recipient) + require.NoError(t, err) + require.Equal(t, before, f.Stats(), "one failing entry cannot cause repeated or unrelated reads") + *now = now.Add(o.maxAge) + _, err = c.SenderInfo(t.Context(), payer, recipient) + require.ErrorIs(t, err, ErrObservationUnavailable) + f.Update(func(s *test.ChainState) { s.FailPayer = common.Address{} }) + advanceObserver(f, *now) + require.NoError(t, o.refresh(t.Context())) + _, err = c.SenderInfo(t.Context(), bad, recipient) + require.NoError(t, err) +} + +func TestPayerCacheEventInvalidation(t *testing.T) { + for _, event := range []string{"DepositFunded", "ReserveFunded", "ReserveClaimed", "Withdrawal", "WinningTicketTransfer", "Unlock", "UnlockCancelled", "round", "activation", "registry-target", "gap", "removed", "logs-failure", "range-unchanged", "range-funded"} { + t.Run(event, func(t *testing.T) { + f, o, now := newTestObserver(t) + payer, other, recipient := common.HexToAddress("0x11"), common.HexToAddress("0x12"), common.HexToAddress("0x22") + initial := warmPayer(t, o, payer, recipient) + _, err := o.senderInfo(t.Context(), other, recipient) + require.NoError(t, err) + ranged := event == "range-unchanged" || event == "range-funded" + steps := 1 + if ranged { + steps = 20 + } + for range steps { + advanceObserver(f, *now) + } + f.Update(func(s *test.ChainState) { + if event != "range-unchanged" { + s.Deposit, s.Claimable, s.WithdrawRound = 13, 17, 9 + } + log := types.Log{Address: s.Contracts["TicketBroker"], BlockHash: s.Hash, BlockNumber: s.Number, Topics: []common.Hash{{}, common.BytesToHash(payer.Bytes())}} + switch event { + case "round": + s.CurrentRound++ // Before lastInitializedRound advances. + case "activation": + log.Address, log.Topics[0] = s.Contracts["BondingManager"], o.chain.Contracts.abis["bondingManager"].Events["TranscoderActivated"].ID + case "registry-target": + log.Address, log.Topics[0] = f.Controller, o.chain.Contracts.abis["controller"].Events["SetContractInfo"].ID + case "gap": + s.Number += 129 + case "removed": + log.Removed = true + case "logs-failure": + s.Fail = "eth_getLogs" + case "range-unchanged", "range-funded": + log.Topics[0] = o.chain.Contracts.abis["ticketBroker"].Events["DepositFunded"].ID + default: + log.Topics[0] = o.chain.Contracts.abis["ticketBroker"].Events[event].ID + } + if event != "range-unchanged" { + s.Logs = []types.Log{log} + } + if event != "logs-failure" { + s.LogsBloom = payerLogsBloom(s.Logs) + } + }) + require.NoError(t, o.refresh(t.Context())) + if ranged { + require.NoError(t, initial.Snapshot.Validate(), "verified continuity preserves the generation") + } + before := f.Stats().Calls["TicketBroker.getSenderInfo"] + info, err := o.senderInfo(t.Context(), payer, recipient) + require.NoError(t, err) + if event == "range-unchanged" { + require.Equal(t, initial.Deposit, info.Deposit) + require.Equal(t, initial.Reserve, info.Reserve) + } else { + require.Equal(t, int64(13), info.Deposit.Int64()) + require.Equal(t, int64(17), info.Reserve.Int64()) + require.Equal(t, int64(9), info.WithdrawRound.Int64()) + before++ + } + require.Equal(t, before, f.Stats().Calls["TicketBroker.getSenderInfo"]) + require.Equal(t, f.State().Number, info.Snapshot.Number.Uint64()) + _, err = o.senderInfo(t.Context(), other, recipient) + require.NoError(t, err) + if _, targeted := o.chain.Contracts.abis["ticketBroker"].Events[event]; targeted || ranged { + require.Equal(t, before, f.Stats().Calls["TicketBroker.getSenderInfo"], "unaffected payer stays warm") + } + before = f.Stats().Calls["TicketBroker.getSenderInfo"] + _, err = o.senderInfo(t.Context(), payer, recipient) + require.NoError(t, err) + require.Equal(t, before, f.Stats().Calls["TicketBroker.getSenderInfo"]) + }) + } +} + +type payerGate struct { + next http.RoundTripper + entered, release chan struct{} + active atomic.Int32 + once bool + used atomic.Bool +} + +func (g *payerGate) RoundTrip(r *http.Request) (*http.Response, error) { + if g.once && g.used.Swap(true) { + return g.next.RoundTrip(r) + } + response, err := g.next.RoundTrip(r) + if err != nil { + return nil, err + } + g.active.Add(1) + defer g.active.Add(-1) + g.entered <- struct{}{} + <-g.release // Hold even a cancelled transport until its actual exit. + if err := r.Context().Err(); err != nil { + response.Body.Close() + return nil, err + } + return response, nil +} + +func TestPayerColdCompletionCannotRestoreInvalidatedFunds(t *testing.T) { + for _, name := range []string{"unchanged", "funded", "registry-target"} { + t.Run(name, func(t *testing.T) { + changed := name != "unchanged" + f, o, now := newTestObserver(t) + require.NoError(t, o.refresh(t.Context())) + g := gatePayerReads(o) + g.once = true + payer := common.HexToAddress("0x11") + done := make(chan error, 1) + go func() { _, err := o.senderInfo(t.Context(), payer, common.Address{}); done <- err }() + <-g.entered + advanceObserver(f, *now) + if changed { + f.Update(func(s *test.ChainState) { + s.Deposit = 0 + s.Logs = []types.Log{{Address: s.Contracts["TicketBroker"], BlockHash: s.Hash, BlockNumber: s.Number, Topics: []common.Hash{o.chain.Contracts.abis["ticketBroker"].Events["DepositFunded"].ID, common.BytesToHash(payer.Bytes())}}} + if name == "registry-target" { + s.Logs[0].Address, s.Logs[0].Topics[0] = f.Controller, o.chain.Contracts.abis["controller"].Events["SetContractInfo"].ID + } + }) + } + require.NoError(t, o.refresh(t.Context())) + require.Contains(t, f.Stats().LogAddresses, f.Controller) + close(g.release) + err := <-done + if changed { + require.ErrorIs(t, err, ErrObservationUnavailable) + } else { + require.NoError(t, err) + } + info, err := o.senderInfo(t.Context(), payer, common.Address{}) + require.NoError(t, err) + require.Equal(t, f.State().Number, info.Snapshot.Number.Uint64()) + if changed { + require.Zero(t, info.Deposit.Sign()) + } + }) + } +} + +func TestPayerColdReadRecoversAtSameHead(t *testing.T) { + for _, timing := range []string{"epoch", "rpc-before", "rpc-during", "rpc-after"} { + t.Run(timing, func(t *testing.T) { + f, o, _ := newTestObserver(t) + require.NoError(t, o.refresh(t.Context())) + head := f.State() + if timing != "epoch" { + f.Update(func(s *test.ChainState) { s.Fail = "TicketBroker.getSenderInfo" }) + } + g := gatePayerReads(o) + g.once = true + payer := common.HexToAddress("0x11") + done := make(chan error, 1) + go func() { _, err := o.senderInfo(t.Context(), payer, common.Address{}); done <- err }() + <-g.entered + release := func() { + close(g.release) + err := <-done + require.Error(t, err) + if timing == "epoch" { + require.ErrorIs(t, err, ErrObservationUnavailable) + } + } + if timing == "rpc-before" { + release() + before := f.Stats() + _, err := o.senderInfo(t.Context(), payer, common.Address{}) + require.Error(t, err) + require.Equal(t, before, f.Stats(), "same-generation failures suppress repeated reads") + } + f.Update(func(s *test.ChainState) { s.Number--; s.Hash = common.HexToHash("0x10") }) + require.ErrorIs(t, o.refresh(t.Context()), ErrObservationUnavailable) + if timing == "epoch" || timing == "rpc-during" { + release() + } + f.Update(func(s *test.ChainState) { *s = head }) + require.NoError(t, o.refresh(t.Context())) + if timing == "rpc-after" { + release() + } + _, err := o.senderInfo(t.Context(), payer, common.Address{}) + require.NoError(t, err, "a failure from an old epoch must not poison a restored-head payer") + }) + } +} + +func gatePayerReads(o *PaymentObserver) *payerGate { + client := o.chain.Contracts.RPC.http + g := &payerGate{next: client.Transport, entered: make(chan struct{}, 16), release: make(chan struct{})} + client.Transport = g + return g +} + +func TestPayerColdReadsBoundDistinctCancellation(t *testing.T) { + f, o, _ := newTestObserver(t) + warm := common.HexToAddress("0xffff") + warmPayer(t, o, warm, common.Address{}) + g := gatePayerReads(o) + defer func() { close(g.release); o.payerReads.Wait() }() + for n := range payerReadLimit { + ctx, cancel := context.WithCancel(t.Context()) + done := make(chan error, 1) + go func() { + _, err := o.senderInfo(ctx, common.BigToAddress(big.NewInt(int64(n+1))), common.Address{}) + done <- err + }() + select { + case <-g.entered: + case <-time.After(time.Second): + t.Fatal("cold read did not start") + } + cancel() + require.ErrorIs(t, <-done, context.Canceled) + } + before := f.Stats() + for n := payerReadLimit + 1; n <= payerEntries+payerReadLimit+1; n++ { + _, err := o.senderInfo(t.Context(), common.BigToAddress(big.NewInt(int64(n))), common.Address{}) + require.ErrorIs(t, err, ErrObservationUnavailable) + } + require.Equal(t, payerReadLimit+1, len(o.payers), "rejected keys must not grow or evict the cache") + _, err := o.senderInfo(t.Context(), warm, common.Address{}) + require.NoError(t, err) + require.Equal(t, before, f.Stats(), "a warm payer remains available without RPCs at saturation") + require.Equal(t, int32(payerReadLimit), g.active.Load()) + o.mu.RLock() + require.Equal(t, payerReadLimit, o.activeReads, "cancelled transports still own their slots") + o.mu.RUnlock() +} + +func TestPayerColdReadSharedWaiters(t *testing.T) { + _, o, _ := newTestObserver(t) + require.NoError(t, o.refresh(t.Context())) + g := gatePayerReads(o) + ctx, cancel := context.WithCancel(t.Context()) + done := make(chan error, 2) + call := func(ctx context.Context) { + _, err := o.senderInfo(ctx, common.HexToAddress("0x11"), common.Address{}) + done <- err + } + go call(ctx) + <-g.entered + go call(t.Context()) + require.Eventually(t, func() bool { + o.mu.RLock() + defer o.mu.RUnlock() + return o.payers[payerKey{payer: common.HexToAddress("0x11")}].job.waiters == 2 + }, time.Second, time.Millisecond) + cancel() + require.ErrorIs(t, <-done, context.Canceled) + close(g.release) + require.NoError(t, <-done) + o.payerReads.Wait() + require.Zero(t, o.activeReads) +} + +func TestPayerCacheEvictionAndIdle(t *testing.T) { + f, o, now := newTestObserver(t) + require.NoError(t, o.refresh(t.Context())) + for n := 1; n <= payerEntries+1; n++ { + *now = now.Add(time.Millisecond) + _, err := o.senderInfo(t.Context(), common.BigToAddress(big.NewInt(int64(n))), common.Address{}) + require.NoError(t, err) + } + require.Len(t, o.payers, payerEntries) + require.NotContains(t, o.payers, payerKey{payer: common.BigToAddress(big.NewInt(1))}) + *now = now.Add(payerIdle) + advanceObserver(f, *now) + require.NoError(t, o.refresh(t.Context())) + _, err := o.senderInfo(t.Context(), common.HexToAddress("0xffff"), common.Address{}) + require.NoError(t, err) + require.Len(t, o.payers, 1) +} + +func TestPayerShutdownJoinsColdReads(t *testing.T) { + _, o, _ := newTestObserver(t) + ctx, cancel := context.WithCancel(t.Context()) + stopped := make(chan struct{}) + go func() { o.Run(ctx); close(stopped) }() + require.Eventually(t, func() bool { _, err := o.Snapshot(t.Context()); return err == nil }, time.Second, time.Millisecond) + g := gatePayerReads(o) + returned := make(chan error, 1) + go func() { + _, err := o.senderInfo(t.Context(), common.HexToAddress("0x11"), common.Address{}) + returned <- err + }() + <-g.entered + cancel() + select { + case <-stopped: + t.Fatal("observer stopped before its cold RPC exited") + default: + } + close(g.release) + select { + case <-stopped: + case <-time.After(time.Second): + t.Fatal("cold read outlived observer shutdown") + } + require.Error(t, <-returned) + require.Zero(t, g.active.Load()) + _, err := o.senderInfo(t.Context(), common.HexToAddress("0x11"), common.Address{}) + require.ErrorIs(t, err, ErrObservationUnavailable) +} + +func TestPayerReorgDuringLogsRejectsCandidate(t *testing.T) { + for _, tc := range []struct { + name string + gap uint64 + fail string + }{{"child", 1, ""}, {"child-log-failure", 1, "eth_getLogs"}, {"range", 20, ""}} { + t.Run(tc.name, func(t *testing.T) { + f, o, now := newTestObserver(t) + o.recipient = common.HexToAddress("0x22") + payer := common.HexToAddress("0x11") + old := warmPayer(t, o, payer, o.recipient).Snapshot + for range tc.gap { + advanceObserver(f, *now) + } + f.Update(func(s *test.ChainState) { + s.Hash, s.Fail = common.HexToHash("0x200"), tc.fail + }) + client := o.chain.Contracts.RPC.http + next := client.Transport + client.Transport = reorgLogTransport{next, func() { + f.Update(func(s *test.ChainState) { + s.Hash, s.Deposit, s.Fail = common.HexToHash("0x201"), 0, "" + s.Logs = []types.Log{{Address: s.Contracts["TicketBroker"], BlockHash: s.Hash, BlockNumber: s.Number, Topics: []common.Hash{o.chain.Contracts.abis["ticketBroker"].Events["Withdrawal"].ID, common.BytesToHash(payer.Bytes())}}} + }) + }} + require.ErrorIs(t, o.refresh(t.Context()), ErrObservationUnavailable) + require.Equal(t, old.BlockHash, o.current.BlockHash, "orphaned candidates publish no state") + require.ErrorIs(t, old.Validate(), ErrObservationUnavailable) + _, err := o.senderInfo(t.Context(), payer, o.recipient) + require.ErrorIs(t, err, ErrObservationUnavailable) + client.Transport = next + require.NoError(t, o.refresh(t.Context())) + info, err := o.senderInfo(t.Context(), payer, o.recipient) + require.NoError(t, err) + require.Zero(t, info.Deposit.Sign(), "recovery reads replacement-chain funds") + require.Equal(t, f.State().Hash, info.Snapshot.BlockHash) + before := f.Stats() + _, err = o.senderInfo(t.Context(), payer, o.recipient) + require.NoError(t, err) + require.Equal(t, before, f.Stats(), "warm reads remain local after recovery") + }) + } +} + +type reorgLogTransport struct { + http.RoundTripper + reorg func() +} + +func (tr reorgLogTransport) RoundTrip(r *http.Request) (*http.Response, error) { + request, err := readPayerHistoryCall(r) + if err != nil { + return nil, err + } + response, err := tr.RoundTripper.RoundTrip(r) + if request.Method == "eth_getLogs" { + // Hash filters may still return retained orphan logs after a reorg. + tr.reorg() + } + return response, err +} + +type payerHistoryRoute func(*http.Request) (*http.Response, error) + +func (fn payerHistoryRoute) RoundTrip(r *http.Request) (*http.Response, error) { return fn(r) } + +type payerHistoryCall struct { + ID json.RawMessage + Method string + Params []json.RawMessage +} + +func readPayerHistoryCall(r *http.Request) (call payerHistoryCall, err error) { + body, err := r.GetBody() + if err != nil { + return call, err + } + defer body.Close() + err = json.NewDecoder(body).Decode(&call) + return call, err +} + +func TestPayerForksAndResets(t *testing.T) { + for _, tc := range []struct { + name string + gap int + displaced, split bool + }{ + {"child", 1, false, false}, + {"gap", 20, false, false}, + {"split-child", 1, false, true}, + {"split-gap", 20, false, true}, + {"complete-proof", 2, true, false}, + {"empty-proof", 2, true, false}, + {"round", 2, true, false}, + {"registry-event", 2, true, false}, + {"logs-error", 2, true, false}, + {"missing-parent", 2, true, false}, + {"beyond-bound", 129, true, false}, + } { + t.Run(tc.name, func(t *testing.T) { + f, o, now := newTestObserver(t) + payer := common.HexToAddress("0x11") + held := warmPayer(t, o, payer, common.Address{}) + fork := test.NewChainRPC(t) + if tc.displaced { + fork.Update(func(s *test.ChainState) { s.Hash, s.Deposit = common.HexToHash("0xb00"), 0 }) + } + for n := 0; n < tc.gap; n++ { + if !tc.displaced { + advanceObserver(f, *now) + f.Update(func(s *test.ChainState) { + s.Deposit = 0 + if n == 0 { + s.Logs = []types.Log{{Address: s.Contracts["TicketBroker"], BlockHash: s.Hash, BlockNumber: s.Number, Topics: []common.Hash{o.chain.Contracts.abis["ticketBroker"].Events["Withdrawal"].ID, common.BytesToHash(payer.Bytes())}}} + } + }) + } + advanceObserver(fork, *now) + fork.Update(func(s *test.ChainState) { + s.Hash = common.BigToHash(new(big.Int).SetUint64(s.Number + 100000)) + if tc.name == "empty-proof" { + s.LogsBloom = types.Bloom{} + } + }) + } + fork.Update(func(s *test.ChainState) { + switch tc.name { + case "round": + s.CurrentRound++ + case "registry-event": + s.Logs = []types.Log{{Address: fork.Controller, BlockHash: s.Hash, BlockNumber: s.Number, Topics: []common.Hash{o.chain.Contracts.abis["controller"].Events["SetContractInfo"].ID}}} + case "logs-error": + s.Fail = "eth_getLogs" + } + }) + endpoint, err := url.Parse(fork.URL) + require.NoError(t, err) + client := o.chain.Contracts.RPC.http + next := client.Transport + revoked := tc.displaced || (tc.split && tc.gap > 1) + validateHeld := func() { + if revoked { + require.ErrorIs(t, held.Snapshot.Validate(), ErrObservationUnavailable) + } else { + require.NoError(t, held.Snapshot.Validate(), "complete canonical ancestry preserves the generation") + } + } + checked := false + client.Transport = payerHistoryRoute(func(r *http.Request) (*http.Response, error) { + request, err := readPayerHistoryCall(r) + if err != nil { + return nil, err + } + if request.Method == "eth_getBlockByNumber" { + var ref string + if err := json.Unmarshal(request.Params[0], &ref); err != nil { + return nil, err + } + if ref == hexutil.EncodeBig(held.Snapshot.Number) || (tc.name == "missing-parent" && ref == hexutil.EncodeBig(new(big.Int).Add(held.Snapshot.Number, big.NewInt(1)))) { + return next.RoundTrip(r) + } + if ref == hexutil.EncodeBig(new(big.Int).Add(held.Snapshot.Number, big.NewInt(int64(tc.gap)))) { + validateHeld() + require.Equal(t, !revoked, o.Status().Healthy, "revocation precedes the next RPC") + checked = true + } + } + if tc.displaced || (tc.split && request.Method == "eth_getLogs") { + r = r.Clone(r.Context()) + r.URL, r.Host = endpoint, endpoint.Host + } + return next.RoundTrip(r) + }) + require.NoError(t, o.refresh(t.Context())) + require.True(t, checked) + validateHeld() + if tc.split { + require.Positive(t, fork.Stats().Calls["eth_getLogs"]) + } + funds, err := o.senderInfo(t.Context(), payer, common.Address{}) + require.NoError(t, err) + require.Zero(t, funds.Deposit.Sign(), "fork-specific history cannot retain pre-withdrawal funds") + canonical := f.State().Hash + if tc.displaced { + canonical = fork.State().Hash + } + require.Equal(t, canonical, funds.Snapshot.BlockHash) + beforeA, beforeB := f.Stats(), fork.Stats() + _, err = o.senderInfo(t.Context(), payer, common.Address{}) + require.NoError(t, err) + require.Equal(t, beforeA, f.Stats(), "recovered warm funds stay local") + require.Equal(t, beforeB, fork.Stats()) + }) + } +} + +func TestPayerHistoryCostAndFreshness(t *testing.T) { + for _, mode := range []string{"empty", "unrelated", "positive", "missing", "null"} { + for _, gap := range []int{1, 20, 128} { + t.Run(fmt.Sprintf("%s/%d", mode, gap), func(t *testing.T) { + f, o, now := newTestObserver(t) + o.recipient = common.HexToAddress("0x22") + payer := common.HexToAddress("0x11") + initial := warmPayer(t, o, payer, o.recipient) + for range gap { + advanceObserver(f, *now) + f.Update(func(s *test.ChainState) { payerHistoryBloom(s, mode) }) + } + f.Update(func(s *test.ChainState) { + s.Before = func(context.Context, string) { *now = now.Add(110 * time.Millisecond) } + }) + f.Reset() + started := *now + require.NoError(t, o.refresh(t.Context())) + stats := f.Stats() + logs := gap + if mode == "empty" || mode == "unrelated" { + logs = 0 + } + require.Equal(t, logs, stats.Calls["eth_getLogs"]) + headers := gap + 1 + if gap > 1 { + headers++ + } // Existing retained-anchor availability check. + require.Equal(t, headers, stats.Calls["eth_getBlockByNumber"]) + require.Equal(t, headers+logs+7, stats.RoundTrips) + require.Equal(t, time.Duration(stats.RoundTrips)*110*time.Millisecond, now.Sub(started)) + before := f.Stats() + funds, err := o.senderInfo(t.Context(), payer, o.recipient) + require.NoError(t, err) + require.Equal(t, initial.Deposit, funds.Deposit) + require.Equal(t, before, f.Stats(), "complete unchanged history retains warm funds") + t.Logf("mode=%s gap=%d total=%d simulated110ms=%s", mode, gap, stats.RoundTrips, now.Sub(started)) + }) + } + } +} + +func TestPayerMalformedBloom(t *testing.T) { + for _, intermediate := range []bool{false, true} { + t.Run(fmt.Sprint(intermediate), func(t *testing.T) { + f, o, now := newTestObserver(t) + payer := common.HexToAddress("0x11") + held := warmPayer(t, o, payer, common.Address{}) + advanceObserver(f, *now) + f.Update(func(s *test.ChainState) { + payerHistoryBloom(s, "malformed") + s.Deposit = 0 + }) + if intermediate { + advanceObserver(f, *now) + f.Update(func(s *test.ChainState) { payerHistoryBloom(s, "empty") }) + require.NoError(t, o.refresh(t.Context())) + require.ErrorIs(t, held.Snapshot.Validate(), ErrObservationUnavailable) + funds, err := o.senderInfo(t.Context(), payer, common.Address{}) + require.NoError(t, err) + require.Zero(t, funds.Deposit.Sign(), "malformed history forces exact funds recovery") + } else { + require.Error(t, o.refresh(t.Context())) + require.Equal(t, held.Snapshot.BlockHash, o.current.BlockHash, "malformed candidate cannot publish") + } + }) + } +} + +func TestPayerIncompleteHistoryResets(t *testing.T) { + for _, mode := range []string{"parent-height", "parent-hash", "log-height", "log-hash"} { + t.Run(mode, func(t *testing.T) { + f, o, now := newTestObserver(t) + payer := common.HexToAddress("0x11") + held := warmPayer(t, o, payer, common.Address{}) + anchor := f.State() + advanceObserver(f, *now) + parent := f.State() + advanceObserver(f, *now) + head := f.State() + method, ref := "eth_getLogs", "" + var result any + if mode == "parent-height" || mode == "parent-hash" { + method, ref = "eth_getBlockByNumber", hexutil.EncodeUint64(parent.Number) + height, hash := parent.Number, parent.Hash + if mode == "parent-height" { + height++ + } else { + hash = anchor.Hash + } + result = map[string]any{"number": hexutil.EncodeUint64(height), "hash": hash, "parentHash": parent.Parent} + } else { + log := types.Log{Address: head.Contracts["TicketBroker"], BlockHash: head.Hash, BlockNumber: head.Number, Topics: []common.Hash{o.chain.Contracts.abis["ticketBroker"].Events["DepositFunded"].ID, common.HexToHash("0x12")}} + if mode == "log-height" { + log.BlockNumber-- + } else { + log.BlockHash = parent.Hash + } + result = []types.Log{log} + } + payload, err := json.Marshal(result) + require.NoError(t, err) + client := o.chain.Contracts.RPC.http + next := client.Transport + client.Transport = payerHistoryRoute(func(r *http.Request) (*http.Response, error) { + request, err := readPayerHistoryCall(r) + if err != nil { + return nil, err + } + if request.Method != method { + return next.RoundTrip(r) + } + if ref != "" { + var requested string + if err := json.Unmarshal(request.Params[0], &requested); err != nil { + return nil, err + } + if requested != ref { + return next.RoundTrip(r) + } + } + data, err := json.Marshal(map[string]any{"jsonrpc": "2.0", "id": request.ID, "result": json.RawMessage(payload)}) + if err != nil { + return nil, err + } + return &http.Response{StatusCode: 200, Header: make(http.Header), Body: io.NopCloser(bytes.NewReader(data)), Request: r}, nil + }) + require.NoError(t, o.refresh(t.Context())) + require.ErrorIs(t, held.Snapshot.Validate(), ErrObservationUnavailable) + before := f.Stats().Calls["TicketBroker.getSenderInfo"] + _, err = o.senderInfo(t.Context(), payer, common.Address{}) + require.NoError(t, err) + require.Equal(t, before+1, f.Stats().Calls["TicketBroker.getSenderInfo"], "incomplete proof must reset even an unaffected payer") + }) + } +} diff --git a/internal/test/chainrpc.go b/internal/test/chainrpc.go index 2f97755..d7abc3b 100644 --- a/internal/test/chainrpc.go +++ b/internal/test/chainrpc.go @@ -8,12 +8,15 @@ import ( "maps" "net/http" "net/http/httptest" + "slices" "strings" "sync" "testing" "time" "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/common/hexutil" + "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/crypto" "github.com/stretchr/testify/require" ) @@ -32,17 +35,21 @@ type ChainRPC struct { } type ChainState struct { - Number, L1Block, Timestamp uint64 - ChainID uint64 - Hash, Parent, RoundHash common.Hash - Round, CurrentRound uint64 - Recipient common.Address - Activation, Deactivation uint64 - OmitL1, OmitTimestamp bool - Fail string - Before func(context.Context, string) - Contracts map[string]common.Address - Results map[ChainCall]string + Number, L1Block, Timestamp uint64 + ChainID uint64 + Hash, Parent, RoundHash common.Hash + Round, CurrentRound uint64 + Recipient common.Address + Activation, Deactivation uint64 + OmitL1, OmitTimestamp bool + Fail string + Before func(context.Context, string) + Contracts map[string]common.Address + Results map[ChainCall]string + Logs []types.Log + LogsBloom any + Deposit, Reserve, Claimable, WithdrawRound uint64 + FailPayer common.Address } type ChainCall struct { @@ -51,14 +58,16 @@ type ChainCall struct { } type ChainRPCStats struct { - RoundTrips int - Calls map[string]int + RoundTrips int + Calls map[string]int + LogAddresses []common.Address } func NewChainRPC(t testing.TB) *ChainRPC { t.Helper() c := &ChainRPC{Controller: common.HexToAddress("0xc001"), stats: ChainRPCStats{Calls: map[string]int{}}} c.state = ChainState{Number: 0x100000, L1Block: 50, Timestamp: uint64(time.Now().Unix()), Hash: common.HexToHash("0xabcd"), RoundHash: common.HexToHash("0x1234"), Round: 5, CurrentRound: 5, Deactivation: ^uint64(0)} + c.state.Deposit, c.state.Reserve, c.state.Claimable = 1e18, 1e18, 1e18 word := func(value string) string { return strings.Repeat("0", 64-len(value)) + value } contractNames := map[string]string{} c.state.Contracts = map[string]common.Address{} @@ -113,6 +122,40 @@ func NewChainRPC(t testing.TB) *ChainRPC { } var result any switch request.Method { + case "eth_getLogs": + var query struct { + BlockHash common.Hash + FromBlock, ToBlock string + Address []common.Address + } + require.NoError(t, json.Unmarshal(request.Params[0], &query)) + from, to := uint64(0), state.Number + if query.BlockHash == (common.Hash{}) { + var err error + from, err = hexutil.DecodeUint64(query.FromBlock) + require.NoError(t, err) + to, err = hexutil.DecodeUint64(query.ToBlock) + require.NoError(t, err) + } else if _, ok := c.stateAt(query.BlockHash.Hex()); !ok { + fail() + return + } + logs := []types.Log{} + c.mu.Lock() + c.stats.LogAddresses = query.Address + for _, frame := range c.history { + for _, log := range frame.Logs { + matches := query.BlockHash == log.BlockHash + if query.BlockHash == (common.Hash{}) { + matches = log.BlockNumber >= from && log.BlockNumber <= to + } + if matches && slices.Contains(query.Address, log.Address) { + logs = append(logs, log) + } + } + } + c.mu.Unlock() + result = logs case "eth_chainId": result = quantity(state.ChainID) case "eth_getBlockByNumber": @@ -126,6 +169,9 @@ func NewChainRPC(t testing.TB) *ChainRPC { } } header := map[string]any{"number": quantity(state.Number), "hash": state.Hash, "parentHash": state.Parent} + if state.LogsBloom != nil { + header["logsBloom"] = state.LogsBloom + } if !state.OmitTimestamp { header["timestamp"] = quantity(state.Timestamp) } @@ -163,9 +209,13 @@ func NewChainRPC(t testing.TB) *ChainRPC { result = "0x" + word("1") } case "TicketBroker.getSenderInfo": - result = "0x" + word("de0b6b3a7640000") + word("0") + word("de0b6b3a7640000") + word("0") + if state.FailPayer != (common.Address{}) && common.HexToAddress(call.Input[10:74]) == state.FailPayer { + fail() + return + } + result = "0x" + word(quantity(state.Deposit)[2:]) + word(quantity(state.WithdrawRound)[2:]) + word(quantity(state.Reserve)[2:]) + word("0") case "TicketBroker.claimableReserve": - result = "0x" + word("de0b6b3a7640000") + result = "0x" + word(quantity(state.Claimable)[2:]) case "TicketBroker.usedTickets": result = "0x" + word("0") default: @@ -222,7 +272,7 @@ func quantity(n uint64) string { return fmt.Sprintf("0x%x", n) } func (c *ChainRPC) Stats() ChainRPCStats { c.mu.Lock() defer c.mu.Unlock() - return ChainRPCStats{RoundTrips: c.stats.RoundTrips, Calls: maps.Clone(c.stats.Calls)} + return ChainRPCStats{RoundTrips: c.stats.RoundTrips, Calls: maps.Clone(c.stats.Calls), LogAddresses: append([]common.Address(nil), c.stats.LogAddresses...)} } func (c *ChainRPC) Reset() { diff --git a/orchestrator/metrics_test.go b/orchestrator/metrics_test.go index 190b4e8..3f6f9c7 100644 --- a/orchestrator/metrics_test.go +++ b/orchestrator/metrics_test.go @@ -350,7 +350,7 @@ func TestUpstreamFailureAfterHeaders(t *testing.T) { } func TestRedemptionMetricsDurableAndIdempotent(t *testing.T) { - _, store, ticket, _, _ := newRecoveryTest(t, 14) + f, store, ticket, _, _ := newRecoveryTest(t, 14) store.metrics = observability.New("orchestrator") store.metrics.Snapshot(store.collectMetrics) store.readiness = &paidReadiness{completed: time.Now(), chain: true, interval: time.Second, snapshot: eth.ChainSnapshot{Block: new(big.Int).Sub(ticket.ParamsExpirationBlock, big.NewInt(1)), Round: big.NewInt(ticket.CreationRound)}} @@ -378,9 +378,9 @@ func TestRedemptionMetricsDurableAndIdempotent(t *testing.T) { require.EqualValues(t, 1, test.MetricSum(t, store.metrics.Registry, "redemption_transitions_total", map[string]string{"event": "awaiting_finality"})) _, err = store.db.Exec(`DROP TRIGGER fail_transition`) require.NoError(t, err) - require.NoError(t, store.finishTicket(t.Context(), ticket.Sig, "used", "")) - require.NoError(t, store.finishTicket(t.Context(), ticket.Sig, "used", "")) - require.NoError(t, store.finishTicket(t.Context(), ticket.Sig, "confirmed", "")) + require.NoError(t, store.finishTicket(t.Context(), ticket.Sig, "used", "", f.collateralSnapshot())) + require.NoError(t, store.finishTicket(t.Context(), ticket.Sig, "used", "", f.collateralSnapshot())) + require.NoError(t, store.finishTicket(t.Context(), ticket.Sig, "confirmed", "", f.collateralSnapshot())) require.EqualValues(t, 1, test.MetricSum(t, store.metrics.Registry, "redemption_settlements_total", nil)) require.Zero(t, test.MetricSum(t, store.metrics.Registry, "redemption_queue_tickets", nil)) store = reopenRecoveryStore(t, store) diff --git a/orchestrator/migrations/004_collateral_watermark.sql b/orchestrator/migrations/004_collateral_watermark.sql new file mode 100644 index 0000000..dbe5f9a --- /dev/null +++ b/orchestrator/migrations/004_collateral_watermark.sql @@ -0,0 +1,7 @@ +-- UP +CREATE TABLE collateral_watermark ( + id INTEGER PRIMARY KEY CHECK(id=1), + number INTEGER NOT NULL CHECK(number>=0), hash TEXT NOT NULL +); +-- DOWN +DROP TABLE collateral_watermark; diff --git a/orchestrator/redeemer.go b/orchestrator/redeemer.go index 8215d5a..227538a 100644 --- a/orchestrator/redeemer.go +++ b/orchestrator/redeemer.go @@ -11,6 +11,7 @@ import ( "time" "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" "github.com/livepeer/node/eth" "github.com/livepeer/node/pm" ) @@ -68,16 +69,16 @@ func RedeemPending(ctx context.Context, store *RedeemerDB, chain eth.PaymentChai if !active { continue } - used, err := chain.IsUsedTicket(ctx, ticket.Hash(), "latest") + used, _, err := chain.TicketUse(ctx, ticket.Hash(), "latest") if err != nil { failures = append(failures, err) continue } if used { - finalized, err := chain.IsUsedTicket(ctx, ticket.Hash(), "finalized") + finalized, observation, err := chain.TicketUse(ctx, ticket.Hash(), "finalized") if err == nil { if finalized { - err = store.setTicketPhase(ctx, ticket.Sig, "used") + err = store.setTicketPhase(ctx, ticket.Sig, "used", observation) } else { err = store.setTicketPhase(ctx, ticket.Sig, "awaiting_finality") } @@ -103,9 +104,9 @@ func RedeemPending(ctx context.Context, store *RedeemerDB, chain eth.PaymentChai func redeemTicket(ticket *pm.SignedTicket) eth.RedeemTicket { return eth.RedeemTicket{Recipient: ticket.Recipient, Sender: ticket.PayerAddress, FaceValue: ticket.FaceValue, WinProb: ticket.WinProb, SenderNonce: ticket.TicketNonce, RecipientRandHash: ticket.RecipientRandHash, AuxData: ticket.AuxData(), Signature: ticket.Sig, RecipientRand: ticket.RecipientRand} } -func (s *RedeemerDB) setTicketPhase(ctx context.Context, sig []byte, phase string) error { +func (s *RedeemerDB) setTicketPhase(ctx context.Context, sig []byte, phase string, observation ...eth.ChainSnapshot) error { if terminalRedemption(phase) { - return s.finishTicket(ctx, sig, phase, "") + return s.finishTicket(ctx, sig, phase, "", observation...) } result, err := s.db.ExecContext(ctx, `INSERT INTO redemption_attempts(sig,attempted_at,phase) VALUES(?,?,?) ON CONFLICT(sig) DO UPDATE SET phase=excluded.phase WHERE redemption_attempts.phase IN ('retryable','awaiting_finality') AND redemption_attempts.phase!=excluded.phase`, sig, time.Now().UTC().Format(time.RFC3339Nano), phase) if err == nil { @@ -117,7 +118,7 @@ func (s *RedeemerDB) setTicketPhase(ctx context.Context, sig []byte, phase strin } type ReceiptReader interface { - Receipt(context.Context, common.Hash) (confirmed, reverted bool, err error) + FinalizedReceipt(context.Context, common.Hash) (*types.Receipt, error) } // ReconcileSubmitted checks every retained identity, including old retry groups. @@ -133,16 +134,16 @@ func ReconcileSubmitted(ctx context.Context, store *RedeemerDB, chain ReceiptRea reverts := map[string]common.Hash{} pending := map[common.Hash]bool{} for _, item := range items { - confirmed, reverted, err := chain.Receipt(ctx, item.Hash) + receipt, err := chain.FinalizedReceipt(ctx, item.Hash) id := string(item.Signature) if err != nil { failures = append(failures, fmt.Errorf("receipt %s: %w", item.Hash.Hex(), err)) continue } - if confirmed { + if receipt != nil && receipt.Status == types.ReceiptStatusSuccessful { successes[id] = true - err = store.finishTicket(ctx, item.Signature, "confirmed", item.Hash.Hex()) - } else if reverted { + err = store.finishTicket(ctx, item.Signature, "confirmed", item.Hash.Hex(), eth.ChainSnapshot{Number: receipt.BlockNumber, BlockHash: receipt.BlockHash}) + } else if receipt != nil && receipt.Status == types.ReceiptStatusFailed { reverts[id] = item.Hash err = store.finishTicket(ctx, item.Signature, "reverted", item.Hash.Hex()) } else { diff --git a/orchestrator/redeemer_db.go b/orchestrator/redeemer_db.go index 5ffb1b0..96c231d 100644 --- a/orchestrator/redeemer_db.go +++ b/orchestrator/redeemer_db.go @@ -135,6 +135,15 @@ func (s *RedeemerDB) storeWinningTickets(ctx context.Context, payer ethcommon.Ad return err } defer tx.Rollback() + // A singleton durable floor prevents old collateral being reused after any + // settlement releases liability. It is deliberately conservative across payers. + var number int64 + var hash string + if err := tx.QueryRowContext(ctx, `SELECT number,hash FROM collateral_watermark WHERE id=1`).Scan(&number, &hash); err != nil && !errors.Is(err, sql.ErrNoRows) { + return err + } else if err == nil && (snapshot.Number == nil || snapshot.Number.Cmp(big.NewInt(number)) < 0 || (snapshot.Number.Cmp(big.NewInt(number)) == 0 && snapshot.BlockHash.Hex() != hash)) { + return pm.ErrPayerUnavailable + } // Once redemption can reveal randomness, this epoch must never accept // more tickets, even if a reorg moves the observed L1 clock backwards. var exposed int diff --git a/orchestrator/redeemer_db_test.go b/orchestrator/redeemer_db_test.go index 26cf461..8689b9c 100644 --- a/orchestrator/redeemer_db_test.go +++ b/orchestrator/redeemer_db_test.go @@ -10,6 +10,7 @@ import ( "time" ethcommon "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" "github.com/livepeer/node/eth" "github.com/livepeer/node/nodeconfig/migrations" "github.com/livepeer/node/pm" @@ -18,8 +19,15 @@ import ( type receiptFixture struct{ confirmed, reverted bool } -func (r receiptFixture) Receipt(context.Context, ethcommon.Hash) (bool, bool, error) { - return r.confirmed, r.reverted, nil +func (r receiptFixture) FinalizedReceipt(context.Context, ethcommon.Hash) (*types.Receipt, error) { + if !r.confirmed && !r.reverted { + return nil, nil + } + status := uint64(types.ReceiptStatusFailed) + if r.confirmed { + status = types.ReceiptStatusSuccessful + } + return &types.Receipt{Status: status, BlockNumber: big.NewInt(10), BlockHash: ethcommon.HexToHash("0x10")}, nil } func TestRedeemerSQLiteFilePermissions(t *testing.T) { @@ -121,6 +129,7 @@ func TestRedemptionMigrationHistoryAndRollback(t *testing.T) { require.NoError(t, err) require.NoError(t, migrations.Down(t.Context(), s.db, files)) require.NoError(t, migrations.Down(t.Context(), s.db, files)) + require.NoError(t, migrations.Down(t.Context(), s.db, files)) // Seed a genuine v1 database, then upgrade without changing its ticket identity. prepared, err := c.PrepareRedemption(t.Context(), key, big.NewInt(1), redeemTicket(ticket)) require.NoError(t, err) @@ -141,6 +150,7 @@ func TestRedemptionMigrationHistoryAndRollback(t *testing.T) { // A lossless rollback remains possible with the original representable state. require.NoError(t, migrations.Down(t.Context(), s.db, files)) require.NoError(t, migrations.Down(t.Context(), s.db, files)) + require.NoError(t, migrations.Down(t.Context(), s.db, files)) require.NoError(t, migrations.Up(t.Context(), s.db, files)) _, err = s.db.Exec("UPDATE redemption_attempts SET phase='retryable' WHERE sig=?", ticket.Sig) require.NoError(t, err) @@ -149,6 +159,7 @@ func TestRedemptionMigrationHistoryAndRollback(t *testing.T) { next.Nonce = 1 require.NoError(t, s.recordPrepared(t.Context(), ticket, next)) require.NoError(t, migrations.Down(t.Context(), s.db, files)) + require.NoError(t, migrations.Down(t.Context(), s.db, files)) require.ErrorContains(t, migrations.Down(t.Context(), s.db, files), "cannot_downgrade_redemption_transaction_history") history, err = s.Redemptions() require.NoError(t, err) @@ -170,6 +181,7 @@ func TestRedemptionRollbackPreservesTransactionProgress(t *testing.T) { files, err := fs.Sub(redeemerMigrationFiles, "migrations") require.NoError(t, err) require.NoError(t, migrations.Down(t.Context(), s.db, files)) + require.NoError(t, migrations.Down(t.Context(), s.db, files)) require.ErrorContains(t, migrations.Down(t.Context(), s.db, files), "cannot_downgrade_redemption_transaction_history") floor, err = s.SavedNonceFloor(t.Context(), key.Address()) require.NoError(t, err) diff --git a/orchestrator/redemption_nonce_test.go b/orchestrator/redemption_nonce_test.go index 4c9a047..e267f13 100644 --- a/orchestrator/redemption_nonce_test.go +++ b/orchestrator/redemption_nonce_test.go @@ -24,7 +24,7 @@ func TestDroppedRedemptionRefillsNonceAfterRestart(t *testing.T) { require.Eventually(t, func() bool { return ticketPhase(t, s, ticket.Sig) == "retryable" }, time.Second, time.Millisecond) s = reopenRecoveryStore(t, s) if phase != "retryable" { - require.NoError(t, s.setTicketPhase(t.Context(), ticket.Sig, phase)) + require.NoError(t, s.setTicketPhase(t.Context(), ticket.Sig, phase, f.collateralSnapshot())) } // A failed claim must prevent both the saved-byte replay and any fresh // higher nonce. Restart retains the original timestamp and identity. @@ -134,7 +134,7 @@ func TestFinalizedExternalNonceRetiresOnlyMissingTransactions(t *testing.T) { require.Empty(t, ReconcileSubmitted(t.Context(), s, c)) next := nextRecoveryTicket(ticket, 235) next.CreationRound = 5 - require.ErrorIs(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, next.RecipientRandHash, 5, big.NewInt(1000), []*pm.SignedTicket{next}, pm.ChainSnapshot{}), pm.ErrPayerUnavailable) + require.ErrorIs(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, next.RecipientRandHash, 5, big.NewInt(1000), []*pm.SignedTicket{next}, f.collateralSnapshot()), pm.ErrPayerUnavailable) f.mu.Lock() f.finalizedNonce = 1 f.finalizedNonceReads = 0 @@ -153,10 +153,10 @@ func TestFinalizedExternalNonceRetiresOnlyMissingTransactions(t *testing.T) { if observation == "missing" { require.Empty(t, items) require.Equal(t, "expired", ticketPhase(t, s, ticket.Sig)) - require.NoError(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, next.RecipientRandHash, 5, big.NewInt(1000), []*pm.SignedTicket{next}, pm.ChainSnapshot{})) + require.NoError(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, next.RecipientRandHash, 5, big.NewInt(1000), []*pm.SignedTicket{next}, f.collateralSnapshot())) oldEpoch := nextRecoveryTicket(ticket, 236) oldEpoch.RecipientRandHash = ticket.RecipientRandHash - require.ErrorIs(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, ticket.RecipientRandHash, 0, big.NewInt(10000), []*pm.SignedTicket{oldEpoch}, pm.ChainSnapshot{}), ErrInvalidPayment) + require.ErrorIs(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, ticket.RecipientRandHash, 0, big.NewInt(10000), []*pm.SignedTicket{oldEpoch}, f.collateralSnapshot()), ErrInvalidPayment) } else { require.Len(t, items, 1, "uncertain or unfinalized receipts retain liability") } @@ -191,7 +191,7 @@ func TestFinalizedExternalNoncePreservesHigherAttempt(t *testing.T) { require.Equal(t, 1, reads) next := nextRecoveryTicket(ticket, 238) next.CreationRound = 5 - require.ErrorIs(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, next.RecipientRandHash, 5, big.NewInt(1000), []*pm.SignedTicket{next}, pm.ChainSnapshot{}), pm.ErrPayerUnavailable) + require.ErrorIs(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, next.RecipientRandHash, 5, big.NewInt(1000), []*pm.SignedTicket{next}, f.collateralSnapshot()), pm.ErrPayerUnavailable) // Finalized state advances beyond both attempts while the pending endpoint // stays stale. New signing must use that observed count, not the saved floor. f.mu.Lock() diff --git a/orchestrator/redemption_recovery.go b/orchestrator/redemption_recovery.go index 9684e9d..bdc584a 100644 --- a/orchestrator/redemption_recovery.go +++ b/orchestrator/redemption_recovery.go @@ -362,17 +362,18 @@ func (r *redemptionRecovery) reconcileUsed(ctx context.Context) []error { failures = append(failures, err) continue } - used, err := r.chain.IsUsedTicket(ctx, ticket.Hash(), "latest") + used, _, err := r.chain.TicketUse(ctx, ticket.Hash(), "latest") if err != nil { failures = append(failures, err) continue } phase := "retryable" + var observation eth.ChainSnapshot if !used && item.phase != "awaiting_finality" { continue } if used { - finalized, err := r.chain.IsUsedTicket(ctx, ticket.Hash(), "finalized") + finalized, evidence, err := r.chain.TicketUse(ctx, ticket.Hash(), "finalized") if err != nil { failures = append(failures, err) continue @@ -380,9 +381,9 @@ func (r *redemptionRecovery) reconcileUsed(ctx context.Context) []error { if !finalized { continue } - phase = "used" + phase, observation = "used", evidence } - if err := r.store.setTicketPhase(ctx, sig, phase); err != nil { + if err := r.store.setTicketPhase(ctx, sig, phase, observation); err != nil { failures = append(failures, err) } } diff --git a/orchestrator/redemption_recovery_test.go b/orchestrator/redemption_recovery_test.go index 8e2f7ec..b3e983b 100644 --- a/orchestrator/redemption_recovery_test.go +++ b/orchestrator/redemption_recovery_test.go @@ -56,6 +56,14 @@ type recoveryRPC struct { server *httptest.Server } +// collateralSnapshot is the same canonical execution identity served by the fixture. +func (f *recoveryRPC) collateralSnapshot() eth.ChainSnapshot { + f.mu.Lock() + defer f.mu.Unlock() + header := &types.Header{Number: big.NewInt(f.block), Difficulty: new(big.Int), BaseFee: big.NewInt(100)} + return eth.ChainSnapshot{Number: header.Number, BlockHash: header.Hash()} +} + func newRecoveryRPC(t *testing.T) *recoveryRPC { t.Helper() f := &recoveryRPC{pool: map[uint64]*types.Transaction{}, receipts: map[common.Hash]*types.Receipt{}, used: map[common.Hash]bool{}, finalized: true, block: 100, round: 1, receiptHeader: &types.Header{Number: big.NewInt(10), Difficulty: new(big.Int), BaseFee: big.NewInt(100)}} @@ -694,11 +702,11 @@ func TestTerminalAttemptRetainsUnresolvedSibling(t *testing.T) { // stops retrying, but cannot release its collateral based on one failed hash. next := nextRecoveryTicket(ticket, 12) next.CreationRound = max(ticket.CreationRound, tc.minRound) - require.ErrorIs(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, next.RecipientRandHash, tc.minRound, big.NewInt(1000), []*pm.SignedTicket{next}, pm.ChainSnapshot{}), pm.ErrPayerUnavailable) + require.ErrorIs(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, next.RecipientRandHash, tc.minRound, big.NewInt(1000), []*pm.SignedTicket{next}, f.collateralSnapshot()), pm.ErrPayerUnavailable) f.mine(sent[0], true) tickRecovery(t, s, c, key) require.Equal(t, "confirmed", ticketPhase(t, s, ticket.Sig)) - require.NoError(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, next.RecipientRandHash, tc.minRound, big.NewInt(1000), []*pm.SignedTicket{next}, pm.ChainSnapshot{})) + require.NoError(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, next.RecipientRandHash, tc.minRound, big.NewInt(1000), []*pm.SignedTicket{next}, f.collateralSnapshot())) }) } } @@ -771,7 +779,7 @@ func TestExpiredAttemptDoesNotReopenRandomness(t *testing.T) { next := nextRecoveryTicket(ticket, 17) next.RecipientRandHash = ticket.RecipientRandHash next.TicketNonce = 2 - require.ErrorIs(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, ticket.RecipientRandHash, 0, big.NewInt(10000), []*pm.SignedTicket{next}, pm.ChainSnapshot{}), ErrInvalidPayment) + require.ErrorIs(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, ticket.RecipientRandHash, 0, big.NewInt(10000), []*pm.SignedTicket{next}, f.collateralSnapshot()), ErrInvalidPayment) } func recoveryReplacement(t *testing.T, key *eth.Key, first *types.Transaction) *types.Transaction { @@ -843,7 +851,7 @@ func TestTerminalTicketKeepsUnresolvedReceipts(t *testing.T) { tickRecovery(t, s, c, key) s = reopenRecoveryStore(t, s) f.mine(f.transactions()[included], true) - require.NoError(t, s.setTicketPhase(t.Context(), ticket.Sig, phase)) + require.NoError(t, s.setTicketPhase(t.Context(), ticket.Sig, phase, f.collateralSnapshot())) require.Empty(t, ReconcileSubmitted(t.Context(), s, c)) require.Equal(t, phase, ticketPhase(t, s, ticket.Sig), "unfinalized inclusion must not settle a terminal ticket") f.mu.Lock() @@ -870,7 +878,7 @@ func TestRevertedReplacementReleasesSameNonceLiability(t *testing.T) { require.NotEmpty(t, ReconcileSubmitted(t.Context(), s, c)) require.Equal(t, "reverted", ticketPhase(t, s, ticket.Sig)) next := nextRecoveryTicket(ticket, 21) - require.NoError(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, next.RecipientRandHash, 0, big.NewInt(1000), []*pm.SignedTicket{next}, pm.ChainSnapshot{}), "a finalized replacement consumes every sibling with the same sender and nonce") + require.NoError(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, next.RecipientRandHash, 0, big.NewInt(1000), []*pm.SignedTicket{next}, f.collateralSnapshot()), "a finalized replacement consumes every sibling with the same sender and nonce") } func TestManualRetryFailureKeepsActiveMonitor(t *testing.T) { @@ -937,12 +945,12 @@ func TestConsumedNonceResolvesOtherTicketHistory(t *testing.T) { require.Equal(t, "retryable", ticketPhase(t, s, sibling.Sig)) require.EqualValues(t, 2, test.MetricSum(t, s.metrics.Registry, "redemption_transitions_total", map[string]string{"event": "retryable"})) require.EqualValues(t, 1, test.MetricSum(t, s.metrics.Registry, "redemption_settlements_total", nil)) - require.NoError(t, s.finishTicket(t.Context(), b.Sig, "confirmed", b0.Hash.Hex())) + require.NoError(t, s.finishTicket(t.Context(), b.Sig, "confirmed", b0.Hash.Hex(), f.collateralSnapshot())) require.EqualValues(t, 2, test.MetricSum(t, s.metrics.Registry, "redemption_transitions_total", map[string]string{"event": "retryable"})) require.EqualValues(t, 1, test.MetricSum(t, s.metrics.Registry, "redemption_settlements_total", nil)) require.NoError(t, s.setTicketPhase(t.Context(), sibling.Sig, "invalid")) next := nextRecoveryTicket(a, 105) - require.NoError(t, s.storeWinningTickets(t.Context(), a.PayerAddress, next.RecipientRandHash, 0, big.NewInt(1000), []*pm.SignedTicket{next}, pm.ChainSnapshot{}), "B consumed nonce zero and A's nonce one failed; A has no remaining liability") + require.NoError(t, s.storeWinningTickets(t.Context(), a.PayerAddress, next.RecipientRandHash, 0, big.NewInt(1000), []*pm.SignedTicket{next}, f.collateralSnapshot()), "B consumed nonce zero and A's nonce one failed; A has no remaining liability") } func TestMultipleFinalizedFailuresResolveEveryGroup(t *testing.T) { @@ -973,7 +981,7 @@ func TestMultipleFinalizedFailuresResolveEveryGroup(t *testing.T) { require.NoError(t, err) require.Len(t, history, 4) next := nextRecoveryTicket(ticket, 107) - require.NoError(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, next.RecipientRandHash, 0, big.NewInt(1000), []*pm.SignedTicket{next}, pm.ChainSnapshot{}), "both finalized failed nonces must resolve their replacement siblings") + require.NoError(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, next.RecipientRandHash, 0, big.NewInt(1000), []*pm.SignedTicket{next}, f.collateralSnapshot()), "both finalized failed nonces must resolve their replacement siblings") } func TestLiveAcknowledgementRestoresNonce(t *testing.T) { @@ -1038,7 +1046,7 @@ func TestUsedTransitionRecoversAfterStorageFailure(t *testing.T) { t.Errorf("summary=%s but redeemed_at remains NULL even after restart and retry", ticketPhase(t, s, ticket.Sig)) } next := nextRecoveryTicket(ticket, 223) - require.NoError(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, next.RecipientRandHash, 0, big.NewInt(1000), []*pm.SignedTicket{next}, pm.ChainSnapshot{}), "a finalized-used ticket must not strand collateral") + require.NoError(t, s.storeWinningTickets(t.Context(), ticket.PayerAddress, next.RecipientRandHash, 0, big.NewInt(1000), []*pm.SignedTicket{next}, f.collateralSnapshot()), "a finalized-used ticket must not strand collateral") } func TestRejectedReplacementKeepsPredecessorReconcilable(t *testing.T) { @@ -1178,3 +1186,37 @@ func TestRedemptionCallbackObservations(t *testing.T) { }) } } + +func TestCollateralWatermarkSurvivesSettlementAndRestart(t *testing.T) { + f, store, ticket, _, _ := newRecoveryTest(t, 243) + captured := f.collateralSnapshot() + settled := eth.ChainSnapshot{Number: new(big.Int).Add(captured.Number, big.NewInt(1)), BlockHash: common.HexToHash("0x101")} + next := nextRecoveryTicket(ticket, 244) + admit := func(snapshot eth.ChainSnapshot) error { + return store.storeWinningTickets(t.Context(), ticket.PayerAddress, next.RecipientRandHash, 0, new(big.Int).Set(ticket.FaceValue), []*pm.SignedTicket{next}, snapshot) + } + require.ErrorContains(t, store.finishTicket(t.Context(), ticket.Sig, "used", ""), "block evidence required") + _, err := store.db.Exec(`CREATE TRIGGER fail_settlement BEFORE UPDATE OF redeemed_at ON winning_tickets BEGIN SELECT RAISE(ABORT,'storage failure'); END`) + require.NoError(t, err) + require.Error(t, store.finishTicket(t.Context(), ticket.Sig, "used", "", settled)) + var floors, redeemed int + require.NoError(t, store.db.QueryRow(`SELECT count(*) FROM collateral_watermark`).Scan(&floors)) + require.Zero(t, floors, "failed settlement cannot advance its watermark") + require.NoError(t, store.db.QueryRow(`SELECT count(*) FROM winning_tickets WHERE redeemed_at IS NOT NULL`).Scan(&redeemed)) + require.Zero(t, redeemed) + require.ErrorIs(t, admit(captured), pm.ErrPayerUnavailable, "failed settlement retains liability") + _, err = store.db.Exec(`DROP TRIGGER fail_settlement`) + require.NoError(t, err) + // The funds observation was captured while the old winner still reserved its + // face value. Settlement can release it before this admission transaction starts. + require.NoError(t, store.finishTicket(t.Context(), ticket.Sig, "used", "", settled)) + require.ErrorIs(t, admit(captured), pm.ErrPayerUnavailable) + store = reopenRecoveryStore(t, store) + require.ErrorIs(t, admit(captured), pm.ErrPayerUnavailable) + wrongFork := settled + wrongFork.BlockHash = common.HexToHash("0xbad") + require.ErrorIs(t, admit(wrongFork), pm.ErrPayerUnavailable) + require.ErrorContains(t, store.finishTicket(t.Context(), ticket.Sig, "used", "", wrongFork), "conflicting finalized") + require.NoError(t, admit(settled), "post-settlement collateral can cover the next winner") + require.Error(t, admit(settled), "durable settlement must retain signature replay protection") +} diff --git a/orchestrator/redemption_store.go b/orchestrator/redemption_store.go index 0207963..adfeb2a 100644 --- a/orchestrator/redemption_store.go +++ b/orchestrator/redemption_store.go @@ -227,12 +227,29 @@ func signedIdentity(tx *types.Transaction) (eth.SignedTransaction, error) { } // finishTicket never discards transaction history or the epoch exposure marker. -func (s *RedeemerDB) finishTicket(ctx context.Context, sig []byte, phase string, hash string) error { +func (s *RedeemerDB) finishTicket(ctx context.Context, sig []byte, phase string, hash string, observation ...eth.ChainSnapshot) error { tx, err := s.db.BeginTx(ctx, nil) if err != nil { return err } defer tx.Rollback() + if phase == "confirmed" || phase == "used" { + if len(observation) != 1 || observation[0].Number == nil || !observation[0].Number.IsInt64() || observation[0].Number.Sign() < 0 || observation[0].BlockHash == (ethcommon.Hash{}) { + return errors.New("settlement block evidence required") + } + block := observation[0] + if _, err := tx.ExecContext(ctx, `INSERT INTO collateral_watermark(id,number,hash) VALUES(1,?,?) ON CONFLICT(id) DO UPDATE SET number=excluded.number,hash=excluded.hash WHERE number 0) }, 3*time.Second, 10*time.Millisecond) require.Equal(t, 200, get("/healthz")) return } require.Eventually(t, func() bool { return get("/readyz") == 200 }, 3*time.Second, 10*time.Millisecond) - for _, state := range []int32{1, 2, 3, 4, 0} { - failure.Store(state) - want := 503 - if state == 0 { - want = 200 - } - require.Equal(t, want, get("/readyz")) - require.Equal(t, 200, get("/healthz")) - } + failure.Store(1) + require.Equal(t, 200, get("/readyz"), "warm readiness retains a fresh observation during an RPC outage") + require.Equal(t, 200, get("/healthz")) + failure.Store(0) _, err := outbox.enqueue(t.Context(), testEvent()) require.NoError(t, err) require.Equal(t, 503, get("/readyz"), "a full outbox must make the signer unready")