diff --git a/eth/contracts_test.go b/eth/contracts_test.go index 21acec7..6f6e986 100644 --- a/eth/contracts_test.go +++ b/eth/contracts_test.go @@ -79,7 +79,7 @@ func TestRedemptionSubmission(t *testing.T) { require.Equal(t, sentHash, hash) // Redemption resolves the live broker independently of an uninitialized observer. broker = common.HexToAddress("0x2001") - chain := PaymentChain{Contracts: contracts, Observer: NewPaymentObserver(PaymentChain{Contracts: contracts}, false)} + chain := PaymentChain{Contracts: contracts, Observer: NewPaymentObserver(PaymentChain{Contracts: contracts}, false, common.Address{})} hash, err = chain.Redeem(t.Context(), key, big.NewInt(1), ticket) require.NoError(t, err) require.Equal(t, sentHash, hash) diff --git a/eth/payment.go b/eth/payment.go index 8d55615..07273ba 100644 --- a/eth/payment.go +++ b/eth/payment.go @@ -30,9 +30,18 @@ func (c PaymentChain) IsActive(ctx context.Context, recipient ethcommon.Address) } func (c PaymentChain) IsActiveAt(ctx context.Context, recipient ethcommon.Address, snapshot ChainSnapshot) (bool, error) { + if err := ctx.Err(); err != nil { + return false, err + } if err := snapshot.Validate(); err != nil { return false, err } + if c.Observer != nil || snapshot.observer != nil { + if snapshot.observer != c.Observer || c.Contracts != c.Observer.chain.Contracts || recipient == (ethcommon.Address{}) || recipient != c.Observer.recipient { + return false, ErrObservationUnavailable + } + return snapshot.recipientActive, snapshot.recipientErr + } if snapshot.blockReference == nil { return false, errors.New("canonical snapshot required") } @@ -40,15 +49,15 @@ func (c PaymentChain) IsActiveAt(ctx context.Context, recipient ethcommon.Addres if err != nil { return false, err } + return c.activeAt(ctx, address, recipient, snapshot) +} + +func (c PaymentChain) activeAt(ctx context.Context, address, recipient ethcommon.Address, snapshot ChainSnapshot) (bool, error) { bonding, err := contracts.NewBondingManagerCaller(address, c.Contracts.callerAt(snapshot.blockReference)) if err != nil { return false, err } - active, err := bonding.IsActiveTranscoder(&bind.CallOpts{Context: ctx}, recipient) - if err == nil { - err = snapshot.Validate() - } - return active, err + return bonding.IsActiveTranscoder(&bind.CallOpts{Context: ctx}, recipient) } // Receipt reports a submitted redemption without retrying or resubmitting it. @@ -81,16 +90,18 @@ func (c PaymentChain) Receipt(ctx context.Context, hash ethcommon.Hash) (confirm } type ChainSnapshot struct { - Block *big.Int // Protocol L1 block, including on Arbitrum. - Round *big.Int // Last initialized round. - RoundHash ethcommon.Hash - CurrentRound *big.Int // Protocol round; may exceed the initialized Round. - blockReference map[string]any - addresses paymentContracts - observer *PaymentObserver - generation uint64 - observedAt time.Time - blockTime time.Time + Block *big.Int // Protocol L1 block, including on Arbitrum. + Round *big.Int // Last initialized round. + RoundHash ethcommon.Hash + CurrentRound *big.Int // Protocol round; may exceed the initialized Round. + blockReference map[string]any + addresses paymentContracts + observer *PaymentObserver + generation uint64 + observedAt time.Time + blockTime time.Time + recipientActive bool + recipientErr error } func (c PaymentChain) Snapshot(ctx context.Context) (ChainSnapshot, error) { diff --git a/eth/payment_contracts_test.go b/eth/payment_contracts_test.go index 218afca..98126ef 100644 --- a/eth/payment_contracts_test.go +++ b/eth/payment_contracts_test.go @@ -20,6 +20,7 @@ func TestObserverConfirmedPaymentContracts(t *testing.T) { }{{"RoundsManager", 5}, {"TicketBroker", 5}, {"BondingManager", 5}, {"RoundsManager", 9}} { t.Run(tc.contract+big.NewInt(int64(tc.round)).String(), func(t *testing.T) { f, o, now := newTestObserver(t) + o.recipient = common.HexToAddress("0x3000") require.NoError(t, o.refresh(t.Context())) before, err := o.Snapshot(t.Context()) require.NoError(t, err) @@ -68,6 +69,7 @@ func TestObserverConfirmedPaymentContracts(t *testing.T) { } } require.Zero(t, f.Stats().Calls["Controller.getContract"], "warm payment reads retain confirmed addresses") + require.Zero(t, f.Stats().Calls["BondingManager.isActiveTranscoder"], "replacement activity belongs to the supplied observation") }) } } @@ -121,7 +123,7 @@ func TestObserverRegistryScopeAndFinalizedReads(t *testing.T) { require.NoError(t, err) f.Reset() _, err = (PaymentChain{Contracts: other}).IsActiveAt(t.Context(), common.HexToAddress("0x3000"), s) - require.ErrorContains(t, err, "another contract client") + require.ErrorIs(t, err, ErrObservationUnavailable) require.Zero(t, f.Stats().RoundTrips) advanceObserver(f, *now) f.Update(func(state *test.ChainState) { diff --git a/eth/payment_observer.go b/eth/payment_observer.go index 44e5047..2baf7b6 100644 --- a/eth/payment_observer.go +++ b/eth/payment_observer.go @@ -34,11 +34,12 @@ type PaymentObserver struct { contact, progress time.Time l1Progress time.Time l1HighWater *big.Int + recipient common.Address } -func NewPaymentObserver(chain PaymentChain, requireL1 bool) *PaymentObserver { +func NewPaymentObserver(chain PaymentChain, requireL1 bool, recipient common.Address) *PaymentObserver { chain.Observer = nil - return &PaymentObserver{chain: chain, interval: 5 * time.Second, maxAge: 30 * time.Second, now: time.Now, requireL1: requireL1 || chain.Contracts.RPC.requireL1} + return &PaymentObserver{chain: chain, interval: 5 * time.Second, maxAge: 30 * time.Second, now: time.Now, requireL1: requireL1 || chain.Contracts.RPC.requireL1, recipient: recipient} } // Run has one service-owned caller. Every RPC inherits the service lifetime. @@ -62,13 +63,14 @@ func (o *PaymentObserver) Run(ctx context.Context) { // A single provider cannot establish an absolute bound on protocol L1 lag. type ObserverStatus struct { ObservedAt, ContactAt, HeadProgressAt, L1ProgressAt time.Time - Healthy bool + Healthy, RecipientActive bool } func (o *PaymentObserver) Status() ObserverStatus { o.mu.RLock() defer o.mu.RUnlock() - return ObserverStatus{o.current.observedAt, o.contact, o.progress, o.l1Progress, o.validate(o.current) == nil} + 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} } func (o *PaymentObserver) Snapshot(ctx context.Context) (ChainSnapshot, error) { @@ -180,6 +182,14 @@ func (o *PaymentObserver) refresh(ctx context.Context) error { if err != nil { return err } + // Local admission follows go-livepeer/core/orchestrator.go isActive at + // 597f882c (Nico Vergauwen, 5bfee8f7; MIT, see LICENSE). Retain node's + // canonical contract predicate rather than duplicating round arithmetic. + // Per-entry failure isolation adapts Mike Zupper's PR #1 eth/chainview.go, + // 68c7834921c7218d16dbd883e2b7e87f6d7804d5 (MIT, see LICENSE). + if o.recipient != (common.Address{}) { + s.recipientActive, s.recipientErr = o.chain.activeAt(ctx, s.addresses.bonding, o.recipient, s) + } if err := ctx.Err(); err != nil { return err } diff --git a/eth/payment_observer_test.go b/eth/payment_observer_test.go index 1a1cd69..f133b08 100644 --- a/eth/payment_observer_test.go +++ b/eth/payment_observer_test.go @@ -25,7 +25,7 @@ func newTestObserver(t *testing.T) (*test.ChainRPC, *PaymentObserver, *time.Time t.Cleanup(rpc.Close) c, err := NewContracts(rpc, f.Controller) require.NoError(t, err) - o := NewPaymentObserver(PaymentChain{Contracts: c}, true) + o := NewPaymentObserver(PaymentChain{Contracts: c}, true, common.Address{}) now := time.Now().Truncate(time.Second) o.now = func() time.Time { return now } return f, o, &now @@ -69,13 +69,13 @@ func TestObserverExplicitArbitrumChainIDRequiresL1(t *testing.T) { f.Update(func(s *test.ChainState) { s.ChainID, s.OmitL1 = 42161, true }) rpc := original.chain.Contracts.RPC require.NoError(t, rpc.CheckChainID(t.Context(), big.NewInt(42161))) - o := NewPaymentObserver(original.chain, false) // No named network preset. + o := NewPaymentObserver(original.chain, false, common.Address{}) // No named network preset. require.ErrorIs(t, o.refresh(t.Context()), ErrObservationUnavailable) // A non-Arbitrum custom chain continues to use its native block clock. f.Update(func(s *test.ChainState) { s.ChainID = 1 }) rpc.requireL1 = false require.NoError(t, rpc.CheckChainID(t.Context(), big.NewInt(1))) - o = NewPaymentObserver(original.chain, false) + o = NewPaymentObserver(original.chain, false, common.Address{}) require.NoError(t, o.refresh(t.Context())) } @@ -337,3 +337,74 @@ func TestObserverGapKeepsCanonicalStateAvailable(t *testing.T) { } } } + +// Round boundary/removal cases adapt Nico Vergauwen's go-livepeer +// eth/watchers/orchestratorwatcher_test.go at 597f882cb9f4d42ad8430b0f592afd5929af4a6b. +// Local/pinned failure isolation adapts Mike Zupper's PR #1 eth/chainview_test.go +// at bbe126e5eddbf59f9cc9f94d3c37703215c26850. MIT notice in LICENSE. +func TestObserverRecipientActivity(t *testing.T) { + f, o, now := newTestObserver(t) + o.recipient = common.HexToAddress("0x1234") + c := PaymentChain{Contracts: o.chain.Contracts, Observer: o} + f.Update(func(s *test.ChainState) { s.Recipient, s.Activation, s.Deactivation = o.recipient, 6, 8 }) + _, err := c.IsActive(t.Context(), o.recipient) + require.ErrorIs(t, err, ErrObservationUnavailable) + for _, step := range []struct { + round uint64 + active bool + }{{5, false}, {6, true}, {7, true}, {8, false}} { + f.Update(func(s *test.ChainState) { s.CurrentRound = step.round }) + require.NoError(t, o.refresh(t.Context())) + snapshot, err := c.Snapshot(t.Context()) + require.NoError(t, err) + require.EqualValues(t, 5, snapshot.Round.Int64(), "activity uses current round even before initialization") + before := f.Stats() + active, err := c.IsActiveAt(t.Context(), o.recipient, snapshot) + require.NoError(t, err) + require.Equal(t, step.active, active) + _, err = c.IsActiveAt(t.Context(), common.HexToAddress("0x9999"), snapshot) + require.ErrorIs(t, err, ErrObservationUnavailable) + unowned := snapshot + unowned.observer = nil + _, err = c.IsActiveAt(t.Context(), o.recipient, unowned) + require.ErrorIs(t, err, ErrObservationUnavailable) + cancelled, cancel := context.WithCancel(t.Context()) + cancel() + _, err = c.IsActiveAt(cancelled, o.recipient, snapshot) + require.ErrorIs(t, err, context.Canceled) + require.Equal(t, before, f.Stats(), "local checks cannot start activity or lookup RPCs") + advanceObserver(f, *now) + f.Update(func(s *test.ChainState) { s.Deactivation = 0 }) + require.NoError(t, o.refresh(t.Context())) + active, err = c.IsActiveAt(t.Context(), o.recipient, snapshot) + require.NoError(t, err) + require.Equal(t, step.active, active, "a newer activity result cannot change a held funds observation") + advanceObserver(f, *now) + f.Update(func(s *test.ChainState) { s.Deactivation = 8 }) + } + + f.Update(func(s *test.ChainState) { s.Fail = "BondingManager.isActiveTranscoder" }) + require.NoError(t, o.refresh(t.Context()), "a recipient failure must not reject unrelated chain state") + require.True(t, o.Status().Healthy) + _, err = c.SenderInfo(t.Context(), common.HexToAddress("0x2000"), o.recipient) + require.NoError(t, err) + _, err = c.IsActive(t.Context(), o.recipient) + require.Error(t, err, "unknown must remain distinct from known inactive") + advanceObserver(f, *now) + f.Update(func(s *test.ChainState) { s.Fail, s.CurrentRound, s.Deactivation = "", 6, 8 }) + require.NoError(t, o.refresh(t.Context())) + old, err := c.Snapshot(t.Context()) + require.NoError(t, err) + f.Update(func(s *test.ChainState) { + s.Hash, s.Parent, s.Deactivation = common.HexToHash("0xdead"), common.Hash{}, 6 + }) + require.NoError(t, o.refresh(t.Context())) + _, err = c.IsActiveAt(t.Context(), o.recipient, old) + require.ErrorIs(t, err, ErrObservationUnavailable) + active, err := c.IsActive(t.Context(), o.recipient) + require.NoError(t, err) + require.False(t, active, "removed activation state is replaced from the canonical block") + *now = now.Add(30 * time.Second) + _, err = c.IsActive(t.Context(), o.recipient) + require.ErrorIs(t, err, ErrObservationUnavailable) +} diff --git a/internal/test/chainrpc.go b/internal/test/chainrpc.go index 91471ee..2f97755 100644 --- a/internal/test/chainrpc.go +++ b/internal/test/chainrpc.go @@ -36,6 +36,8 @@ type ChainState struct { 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) @@ -56,7 +58,7 @@ type ChainRPCStats struct { 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} + 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)} word := func(value string) string { return strings.Repeat("0", 64-len(value)) + value } contractNames := map[string]string{} c.state.Contracts = map[string]common.Address{} @@ -153,7 +155,13 @@ func NewChainRPC(t testing.TB) *ChainRPC { case "RoundsManager.blockHashForRound": result = state.RoundHash.Hex() case "BondingManager.isActiveTranscoder": - result = "0x" + word("1") + if state.Recipient != (common.Address{}) { + require.Equal(t, word(hex.EncodeToString(state.Recipient.Bytes())), call.Input[10:]) + } + result = "0x" + word("0") + if state.Activation <= state.CurrentRound && state.CurrentRound < state.Deactivation { + result = "0x" + word("1") + } case "TicketBroker.getSenderInfo": result = "0x" + word("de0b6b3a7640000") + word("0") + word("de0b6b3a7640000") + word("0") case "TicketBroker.claimableReserve": diff --git a/orchestrator/app.go b/orchestrator/app.go index 2fd5e6b..e15b2d5 100644 --- a/orchestrator/app.go +++ b/orchestrator/app.go @@ -284,7 +284,6 @@ func Serve(parent context.Context, p Params, logOut io.Writer) (result error) { } paymentChain = eth.PaymentChain{Contracts: contracts} observedChain = paymentChain - observedChain.Observer = eth.NewPaymentObserver(paymentChain, p.Network == nodeconfig.Mainnet) if p.ETHUSDFeed != nil { rate, until, err := contracts.WeiPerUSD(parent, *p.ETHUSDFeed, p.PriceMaxAge) if err != nil { @@ -298,6 +297,7 @@ func Serve(parent context.Context, p Params, logOut io.Writer) (result error) { if p.OrchestratorAddress != nil { recipient = *p.OrchestratorAddress } + observedChain.Observer = eth.NewPaymentObserver(paymentChain, p.Network == nodeconfig.Mainnet, recipient) engine, err = NewPaymentEngine(redeemerDB, pm.EthereumChain{Client: observedChain}, recipient, p.TicketFaceValue.ToBig(), p.TicketWinProb.ToBig()) if err != nil { return fmt.Errorf("initialize payment engine (check --account, --ticket-face-value and --ticket-win-prob): %w", err) diff --git a/orchestrator/app_test.go b/orchestrator/app_test.go index a4a595d..dda87bf 100644 --- a/orchestrator/app_test.go +++ b/orchestrator/app_test.go @@ -101,6 +101,9 @@ func TestPaidHTTPStartup(t *testing.T) { result = fmt.Sprintf("0x%064x", 1) var call struct{ Input string } require.NoError(t, json.Unmarshal(req.Params[0], &call)) + if strings.HasPrefix(call.Input, fmt.Sprintf("0x%x", crypto.Keccak256([]byte("isActiveTranscoder(address)"))[:4])) { + require.True(t, strings.HasSuffix(call.Input, fmt.Sprintf("%064x", recipient.Bytes())), "observe the configured ticket recipient, including a separate wallet") + } if strings.HasPrefix(call.Input, fmt.Sprintf("0x%x", crypto.Keccak256([]byte("usedTickets(bytes32)"))[:4])) { result = fmt.Sprintf("0x%064x", 0) } diff --git a/orchestrator/metrics.go b/orchestrator/metrics.go index db9d9af..77a4b6e 100644 --- a/orchestrator/metrics.go +++ b/orchestrator/metrics.go @@ -131,10 +131,11 @@ type paidReadiness struct { func (p *paidReadiness) ready(registry *Registry) bool { p.mu.Lock() limit := max(time.Minute, 3*p.interval) - ok := p.chain && p.active && p.storage && !p.completed.IsZero() && time.Since(p.completed) <= limit && !p.workerCompleted.IsZero() && time.Since(p.workerCompleted) <= limit + chainReady := p.chain && p.active if p.observer != nil { - ok = ok && p.observer.Status().Healthy + chainReady = p.observer.Status().RecipientActive } + ok := chainReady && p.storage && !p.completed.IsZero() && time.Since(p.completed) <= limit && !p.workerCompleted.IsZero() && time.Since(p.workerCompleted) <= limit p.mu.Unlock() registry.mu.Lock() fresh := registry.rateUntil.IsZero() || time.Now().Before(registry.rateUntil) diff --git a/orchestrator/metrics_test.go b/orchestrator/metrics_test.go index 5013daf..190b4e8 100644 --- a/orchestrator/metrics_test.go +++ b/orchestrator/metrics_test.go @@ -40,7 +40,7 @@ func TestPaidReadinessUsesCurrentPublication(t *testing.T) { defer rpc.Close() contracts, err := eth.NewContracts(rpc, f.Controller) require.NoError(t, err) - o := eth.NewPaymentObserver(eth.PaymentChain{Contracts: contracts}, true) + o := eth.NewPaymentObserver(eth.PaymentChain{Contracts: contracts}, true, common.HexToAddress("0x1234")) ctx, cancel := context.WithCancel(t.Context()) done := make(chan struct{}) go func() { o.Run(ctx); close(done) }() diff --git a/orchestrator/payment_recipient_test.go b/orchestrator/payment_recipient_test.go index fc34713..79541e7 100644 --- a/orchestrator/payment_recipient_test.go +++ b/orchestrator/payment_recipient_test.go @@ -471,3 +471,65 @@ func TestLiveChargeObservesOnlyDebitAttempts(t *testing.T) { require.EqualValues(t, 1, test.MetricSum(t, e.metrics.Registry, "payment_operations_total", map[string]string{"operation": "charge", "outcome": "success"})) require.EqualValues(t, 1, test.MetricSum(t, e.metrics.Registry, "payment_operations_total", map[string]string{"operation": "charge", "outcome": "insufficient_balance"})) } + +// Warm admission and refusal cases adapt Mike Zupper's PR #1 +// orchestrator/chainview_test.go at bbe126e5eddbf59f9cc9f94d3c37703215c26850. +// Copyright (c) 2017 Livepeer; MIT notice in LICENSE. +func TestPaymentsAndReadinessUseObservedRecipient(t *testing.T) { + e, _, key, _ := newRecipientTestEngine(t) + f := test.NewChainRPC(t) + f.Update(func(s *test.ChainState) { s.Recipient = e.recipient }) + rpc, err := eth.OpenRPC(f.URL) + require.NoError(t, err) + t.Cleanup(rpc.Close) + contracts, err := eth.NewContracts(rpc, f.Controller) + require.NoError(t, err) + chain := eth.PaymentChain{Contracts: contracts} + chain.Observer = eth.NewPaymentObserver(chain, true, e.recipient) + e.chain = pm.EthereumChain{Client: chain} + ctx, cancel := context.WithCancel(t.Context()) + done := make(chan struct{}) + go func() { chain.Observer.Run(ctx); close(done) }() + t.Cleanup(func() { + cancel() + select { + case <-done: + case <-time.After(time.Second): + t.Error("recipient observation did not stop") + } + }) + require.Eventually(t, func() bool { + _, err := chain.Snapshot(t.Context()) + return err == nil + }, time.Second, time.Millisecond) + registry := NewRegistry("", "https://orch.example", time.Second, time.Minute) + ready := &paidReadiness{storage: true, completed: time.Now(), workerCompleted: time.Now(), observer: chain.Observer} + f.Reset() + info := issueRecipientParams(t, e, key) + require.NoError(t, receiveRecipientPayment(t, e, key, info, 1)) + require.True(t, ready.ready(registry), "current local activity overrides an older unavailable dependency check") + require.Zero(t, f.Stats().Calls["BondingManager.isActiveTranscoder"], "warm challenge, receipt, and readiness activity must be local") + before, err := e.Balance("session") + require.NoError(t, err) + beforeNonces := e.nonceCount + ready.chain, ready.active = true, true // Simulate the preceding periodic check. + f.Update(func(s *test.ChainState) { + s.Parent, s.Hash = s.Hash, ethcommon.HexToHash("0xfeed") + s.Number++ + s.L1Block++ + s.Timestamp = uint64(time.Now().Unix()) + s.Fail = "BondingManager.isActiveTranscoder" + }) + require.Eventually(t, func() bool { + snapshot, err := chain.Snapshot(t.Context()) + return err == nil && snapshot.Block.Int64() == 51 + }, 7*time.Second, 10*time.Millisecond) + f.Reset() + require.False(t, ready.ready(registry), "readiness must refuse between periodic checks") + require.Error(t, receiveRecipientPayment(t, e, key, info, 2)) + after, err := e.Balance("session") + require.NoError(t, err) + require.Equal(t, before.RatString(), after.RatString()) + require.Equal(t, beforeNonces, e.nonceCount, "unavailable activity must not consume replay guards") + require.Zero(t, f.Stats().Calls["BondingManager.isActiveTranscoder"], "unknown state must refuse without a request-triggered retry") +} diff --git a/signer/app.go b/signer/app.go index ba7f978..5a61bce 100644 --- a/signer/app.go +++ b/signer/app.go @@ -186,7 +186,7 @@ func serve(parent context.Context, p Params) error { return fmt.Errorf("initialize signer contract bindings: %w", err) } chain := eth.PaymentChain{Contracts: contracts} - observer := eth.NewPaymentObserver(chain, p.Network == nodeconfig.Mainnet) + observer := eth.NewPaymentObserver(chain, p.Network == nodeconfig.Mainnet, ethcommon.Address{}) chain.Observer = observer service.SetPaymentChain(pm.EthereumChain{Client: chain}) policy, _ := newPricePolicy(p.MaxHourlyPrice, p.MaxFixedPrice) diff --git a/signer/payment_test.go b/signer/payment_test.go index 9317339..96818bb 100644 --- a/signer/payment_test.go +++ b/signer/payment_test.go @@ -227,7 +227,7 @@ func TestPaidRequestDirectRPCBudget(t *testing.T) { info, err = engine.ChallengeInfo(manifest) require.NoError(t, err) for _, during := range []bool{false, true} { - observer := eth.NewPaymentObserver(direct.Client, false) + observer := eth.NewPaymentObserver(direct.Client, false, ethcommon.Address{}) ctx, cancel := context.WithCancel(t.Context()) done := make(chan struct{}) go func() { observer.Run(ctx); close(done) }()