diff --git a/control-plane/cmd/controlplane/main.go b/control-plane/cmd/controlplane/main.go index a5af27c..604e80e 100644 --- a/control-plane/cmd/controlplane/main.go +++ b/control-plane/cmd/controlplane/main.go @@ -120,8 +120,14 @@ func run() error { } server := grpc.NewServer(options...) workloadRepository := workloadapi.NewPostgresRepository(pool) - service := providerjoin.NewService(providerjoin.NewPostgresRepository(pool), providerjoin.NewRedisHeartbeatStore(redisClient), registrar) + providerRepository := providerjoin.NewPostgresRepository(pool) + service := providerjoin.NewService(providerRepository, providerjoin.NewRedisHeartbeatStore(redisClient), registrar) service.SetWorkloadService(workloadapi.NewService(workloadRepository)) + // Independently drives provider_chain_registrations rows left in + // READY/RETRY (e.g. after a Control Plane or chain restart) to + // FINALIZED, without depending on the Agent retrying CompleteJoin. + reconciler := providerjoin.NewReconciler(providerRepository, providerRepository, registrar, providerjoin.DefaultReconcilerConfig()) + go reconciler.Run(ctx) directory := agentmanager.NewDirectory(agentmanager.NewPostgresRegistry(pool), agentmanager.NewRedisLivenessStore(redisClient)) agentClient, err := agentmanager.NewMTLSClient(os.Getenv("AGENT_CLIENT_TLS_CERT_FILE"), os.Getenv("AGENT_CLIENT_TLS_KEY_FILE"), os.Getenv("AGENT_CLIENT_TLS_CA_FILE")) if err != nil { diff --git a/control-plane/internal/providerjoin/postgres.go b/control-plane/internal/providerjoin/postgres.go index 4e345b6..6d918c1 100644 --- a/control-plane/internal/providerjoin/postgres.go +++ b/control-plane/internal/providerjoin/postgres.go @@ -227,6 +227,59 @@ func (r *PostgresRepository) ActivateProvider(ctx context.Context, providerID st return result, nil } +// DueChainRegistrations implements ChainRegistrationStore for the +// Reconciler: providers whose outbox row is READY or RETRY and due (no +// backoff scheduled, or the backoff has elapsed), oldest first so a +// persistently failing registration cannot starve others out of the batch. +func (r *PostgresRepository) DueChainRegistrations(ctx context.Context, limit int) ([]PendingChainRegistration, error) { + rows, err := r.pool.Query(ctx, ` + SELECT cr.provider_id, p.public_key, cr.attempt_count + FROM provider_chain_registrations cr + JOIN providers p ON p.provider_id = cr.provider_id + WHERE cr.state IN ('READY', 'RETRY') + AND (cr.next_attempt_at IS NULL OR cr.next_attempt_at <= now()) + ORDER BY cr.created_at + LIMIT $1`, limit) + if err != nil { + return nil, err + } + defer rows.Close() + + var due []PendingChainRegistration + for rows.Next() { + var pending PendingChainRegistration + if err := rows.Scan(&pending.ProviderID, &pending.PublicKey, &pending.AttemptCount); err != nil { + return nil, err + } + due = append(due, pending) + } + return due, rows.Err() +} + +// RecordChainRegistrationFailure implements ChainRegistrationStore. +func (r *PostgresRepository) RecordChainRegistrationFailure(ctx context.Context, providerID string, attemptErr error, nextAttemptAt time.Time, terminal bool) error { + state := "RETRY" + var next *time.Time + if terminal { + state = "FAILED" + } else { + next = &nextAttemptAt + } + command, err := r.pool.Exec(ctx, ` + UPDATE provider_chain_registrations + SET state = $2, attempt_count = attempt_count + 1, next_attempt_at = $3, + last_error = $4, updated_at = now() + WHERE provider_id = $1`, + providerID, state, next, attemptErr.Error()) + if err != nil { + return err + } + if command.RowsAffected() != 1 { + return ErrProviderNotFound + } + return nil +} + func (r *PostgresRepository) challengeByBeginRequestID(ctx context.Context, requestID string) (Challenge, error) { return scanChallenge(r.pool.QueryRow(ctx, ` SELECT challenge_id, begin_request_id, request_hash, public_key, diff --git a/control-plane/internal/providerjoin/reconciler.go b/control-plane/internal/providerjoin/reconciler.go new file mode 100644 index 0000000..02b1eb1 --- /dev/null +++ b/control-plane/internal/providerjoin/reconciler.go @@ -0,0 +1,199 @@ +package providerjoin + +import ( + "context" + "crypto/ed25519" + "errors" + "log/slog" + "time" +) + +// PendingChainRegistration is one row of the provider_chain_registrations +// outbox that still needs an on-chain registration attempt. +type PendingChainRegistration struct { + ProviderID string + PublicKey []byte + AttemptCount int +} + +// ChainRegistrationStore lets the Reconciler discover and update outbox rows +// independent of any Agent CompleteJoin call. Implemented by +// *PostgresRepository. +type ChainRegistrationStore interface { + // DueChainRegistrations returns up to limit providers whose chain + // registration is READY or RETRY and due (next_attempt_at unset or in + // the past), oldest first. + DueChainRegistrations(ctx context.Context, limit int) ([]PendingChainRegistration, error) + // RecordChainRegistrationFailure increments the attempt counter and + // either schedules the next retry (state RETRY, nextAttemptAt) or, when + // terminal is true, marks the registration FAILED so it stops being + // picked up -- an explicit terminal state rather than a silent hang. + RecordChainRegistrationFailure(ctx context.Context, providerID string, attemptErr error, nextAttemptAt time.Time, terminal bool) error +} + +// Activator is the subset of Repository the Reconciler needs to finalize a +// successful on-chain registration; *PostgresRepository satisfies this via +// its existing ActivateProvider (the same method CompleteJoin's inline path +// already uses). +type Activator interface { + ActivateProvider(ctx context.Context, providerID string, finalization ChainFinalization) (Completion, error) +} + +// ReconcilerConfig bounds the Reconciler's polling cadence, batch size, and +// retry backoff. +type ReconcilerConfig struct { + Interval time.Duration + BatchSize int + MaxAttempts int + BaseBackoff time.Duration + MaxBackoff time.Duration +} + +// DefaultReconcilerConfig returns production-sane bounds: poll every 15s, +// up to 20 providers per pass, exponential backoff from 5s capped at 10m, +// and a registration is declared FAILED (terminal) after 10 attempts. +func DefaultReconcilerConfig() ReconcilerConfig { + return ReconcilerConfig{ + Interval: 15 * time.Second, + BatchSize: 20, + MaxAttempts: 10, + BaseBackoff: 5 * time.Second, + MaxBackoff: 10 * time.Minute, + } +} + +func (c ReconcilerConfig) withDefaults() ReconcilerConfig { + defaults := DefaultReconcilerConfig() + if c.Interval <= 0 { + c.Interval = defaults.Interval + } + if c.BatchSize <= 0 { + c.BatchSize = defaults.BatchSize + } + if c.MaxAttempts <= 0 { + c.MaxAttempts = defaults.MaxAttempts + } + if c.BaseBackoff <= 0 { + c.BaseBackoff = defaults.BaseBackoff + } + if c.MaxBackoff <= 0 { + c.MaxBackoff = defaults.MaxBackoff + } + return c +} + +// Reconciler autonomously drives provider_chain_registrations rows left in +// READY or RETRY to FINALIZED (via the same idempotent EnsureActive/ +// ActivateProvider path CompleteJoin uses) or, after MaxAttempts, to an +// explicit FAILED state. It exists so Provider Join recovers after a +// Control Plane or chain restart without depending on the Agent retrying +// CompleteJoin (issue #10): the outbox row is already committed +// transactionally alongside the provider record in CompleteJoin, so a crash +// between that commit and a successful chain registration leaves work for +// the Reconciler to pick up on its own schedule. +type Reconciler struct { + store ChainRegistrationStore + activator Activator + registrar ProviderRegistrar + now func() time.Time + cfg ReconcilerConfig +} + +func NewReconciler(store ChainRegistrationStore, activator Activator, registrar ProviderRegistrar, cfg ReconcilerConfig) *Reconciler { + return &Reconciler{ + store: store, + activator: activator, + registrar: registrar, + now: time.Now, + cfg: cfg.withDefaults(), + } +} + +// Run polls on cfg.Interval until ctx is cancelled. Intended to be started +// as `go reconciler.Run(ctx)` alongside the gRPC/HTTP servers. +func (r *Reconciler) Run(ctx context.Context) { + ticker := time.NewTicker(r.cfg.Interval) + defer ticker.Stop() + for { + r.ReconcileOnce(ctx) + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + } +} + +// ReconcileOnce processes a single due batch and returns without waiting on +// the ticker -- exported so tests (and any future manual-trigger endpoint) +// can drive reconciliation deterministically. +func (r *Reconciler) ReconcileOnce(ctx context.Context) { + due, err := r.store.DueChainRegistrations(ctx, r.cfg.BatchSize) + if err != nil { + slog.Error("reconciler: failed to list due chain registrations", "error", err) + return + } + for _, registration := range due { + if ctx.Err() != nil { + return + } + r.reconcileOne(ctx, registration) + } +} + +func (r *Reconciler) reconcileOne(ctx context.Context, registration PendingChainRegistration) { + if r.registrar == nil { + return + } + if len(registration.PublicKey) != ed25519.PublicKeySize { + // Malformed stored data: retrying can never fix a wrong-length key. + r.fail(ctx, registration.ProviderID, errors.New("stored public key has an unexpected length"), registration.AttemptCount+1, true) + return + } + var publicKey [ed25519.PublicKeySize]byte + copy(publicKey[:], registration.PublicKey) + + extrinsicHash, blockHash, blockNumber, err := r.registrar.EnsureActive(ctx, publicKey) + if err != nil { + nextAttemptNumber := registration.AttemptCount + 1 + r.fail(ctx, registration.ProviderID, err, nextAttemptNumber, nextAttemptNumber >= r.cfg.MaxAttempts) + return + } + if _, err := r.activator.ActivateProvider(ctx, registration.ProviderID, ChainFinalization{ + ExtrinsicHash: extrinsicHash, + FinalizedBlockHash: blockHash, + FinalizedBlockNumber: blockNumber, + }); err != nil { + slog.Error("reconciler: on-chain registration succeeded but activation failed", + "provider_id", registration.ProviderID, "error", err) + } +} + +func (r *Reconciler) fail(ctx context.Context, providerID string, attemptErr error, attemptNumber int, terminal bool) { + slog.Error("reconciler: chain registration attempt failed", "provider_id", providerID, "error", attemptErr, "attempt", attemptNumber, "terminal", terminal) + nextAttemptAt := r.now().UTC().Add(r.backoffFor(attemptNumber)) + if recErr := r.store.RecordChainRegistrationFailure(ctx, providerID, attemptErr, nextAttemptAt, terminal); recErr != nil { + slog.Error("reconciler: failed to record chain registration attempt", "provider_id", providerID, "error", recErr) + } +} + +// backoffFor returns the base backoff doubled per additional attempt, +// capped at MaxBackoff. attempt is always >= 1. +func (r *Reconciler) backoffFor(attempt int) time.Duration { + const maxShift = 16 // 2^16 * BaseBackoff already exceeds any sane MaxBackoff + shift := attempt - 1 + if shift > maxShift { + shift = maxShift + } + backoff := r.cfg.BaseBackoff + for i := 0; i < shift; i++ { + backoff *= 2 + if backoff <= 0 || backoff > r.cfg.MaxBackoff { + return r.cfg.MaxBackoff + } + } + if backoff > r.cfg.MaxBackoff { + return r.cfg.MaxBackoff + } + return backoff +} diff --git a/control-plane/internal/providerjoin/reconciler_test.go b/control-plane/internal/providerjoin/reconciler_test.go new file mode 100644 index 0000000..9421a02 --- /dev/null +++ b/control-plane/internal/providerjoin/reconciler_test.go @@ -0,0 +1,352 @@ +package providerjoin + +import ( + "context" + "crypto/ed25519" + "errors" + "testing" + "time" +) + +type chainRegistrationRow struct { + publicKey []byte + state string + attemptCount int + nextAttempt time.Time + lastError string +} + +// fakeChainStore is an in-memory ChainRegistrationStore + Activator double. +// Unlike memoryRepository in service_test.go, it only implements what the +// Reconciler needs, and models state transitions explicitly so tests can +// assert on them (state, attempt_count, next_attempt_at, last_error) the +// same way the real Postgres columns would be inspected. +type fakeChainStore struct { + rows map[string]*chainRegistrationRow + activated map[string]ChainFinalization + activateErr error +} + +func newFakeChainStore() *fakeChainStore { + return &fakeChainStore{ + rows: make(map[string]*chainRegistrationRow), + activated: make(map[string]ChainFinalization), + } +} + +func (s *fakeChainStore) addReady(providerID string, publicKey []byte) { + s.rows[providerID] = &chainRegistrationRow{publicKey: publicKey, state: "READY"} +} + +func (s *fakeChainStore) DueChainRegistrations(_ context.Context, limit int) ([]PendingChainRegistration, error) { + var due []PendingChainRegistration + for providerID, row := range s.rows { + if row.state != "READY" && row.state != "RETRY" { + continue + } + if !row.nextAttempt.IsZero() && row.nextAttempt.After(time.Now()) { + continue + } + due = append(due, PendingChainRegistration{ProviderID: providerID, PublicKey: row.publicKey, AttemptCount: row.attemptCount}) + if len(due) == limit { + break + } + } + return due, nil +} + +func (s *fakeChainStore) RecordChainRegistrationFailure(_ context.Context, providerID string, attemptErr error, nextAttemptAt time.Time, terminal bool) error { + row, ok := s.rows[providerID] + if !ok { + return ErrProviderNotFound + } + row.attemptCount++ + row.lastError = attemptErr.Error() + if terminal { + row.state = "FAILED" + row.nextAttempt = time.Time{} + } else { + row.state = "RETRY" + row.nextAttempt = nextAttemptAt + } + return nil +} + +func (s *fakeChainStore) ActivateProvider(_ context.Context, providerID string, finalization ChainFinalization) (Completion, error) { + if s.activateErr != nil { + return Completion{}, s.activateErr + } + row, ok := s.rows[providerID] + if !ok { + return Completion{}, ErrProviderNotFound + } + row.state = "FINALIZED" + s.activated[providerID] = finalization + return Completion{ProviderID: providerID, Status: 2}, nil +} + +// fakeRegistrar lets each test control EnsureActive's outcome per call, +// including failing N times before succeeding to exercise retry/backoff. +type fakeRegistrar struct { + behavior func(calls int) ([]byte, []byte, uint64, error) + calls map[string]int +} + +func newFakeRegistrar(behavior func(calls int) ([]byte, []byte, uint64, error)) *fakeRegistrar { + return &fakeRegistrar{behavior: behavior, calls: make(map[string]int)} +} + +func (r *fakeRegistrar) EnsureActive(_ context.Context, provider [ed25519.PublicKeySize]byte) ([]byte, []byte, uint64, error) { + key := string(provider[:]) + r.calls[key]++ + return r.behavior(r.calls[key]) +} + +func testPublicKey(seed byte) []byte { + key := make([]byte, ed25519.PublicKeySize) + for i := range key { + key[i] = seed + } + return key +} + +func TestReconcileOnceActivatesASuccessfulReadyRegistration(t *testing.T) { + store := newFakeChainStore() + store.addReady("provider-1", testPublicKey(1)) + registrar := newFakeRegistrar(func(int) ([]byte, []byte, uint64, error) { + return []byte("extrinsic"), []byte("block"), 42, nil + }) + reconciler := NewReconciler(store, store, registrar, DefaultReconcilerConfig()) + + reconciler.ReconcileOnce(context.Background()) + + if store.rows["provider-1"].state != "FINALIZED" { + t.Fatalf("expected FINALIZED, got %s", store.rows["provider-1"].state) + } + if finalization, ok := store.activated["provider-1"]; !ok || finalization.FinalizedBlockNumber != 42 { + t.Fatalf("expected ActivateProvider to be called with the registrar's finalization, got %+v", finalization) + } +} + +func TestReconcileOnceSchedulesRetryWithBackoffOnFailure(t *testing.T) { + store := newFakeChainStore() + store.addReady("provider-1", testPublicKey(1)) + failure := errors.New("substrate rpc unavailable") + registrar := newFakeRegistrar(func(int) ([]byte, []byte, uint64, error) { + return nil, nil, 0, failure + }) + cfg := DefaultReconcilerConfig() + cfg.MaxAttempts = 10 + reconciler := NewReconciler(store, store, registrar, cfg) + + before := time.Now() + reconciler.ReconcileOnce(context.Background()) + + row := store.rows["provider-1"] + if row.state != "RETRY" { + t.Fatalf("expected RETRY, got %s", row.state) + } + if row.attemptCount != 1 { + t.Fatalf("expected attempt_count 1, got %d", row.attemptCount) + } + if row.lastError != failure.Error() { + t.Fatalf("expected last_error to be recorded, got %q", row.lastError) + } + if !row.nextAttempt.After(before) { + t.Fatalf("expected next_attempt_at to be scheduled in the future") + } +} + +func TestReconcileMarksFailedAfterMaxAttempts(t *testing.T) { + store := newFakeChainStore() + store.addReady("provider-1", testPublicKey(1)) + failure := errors.New("provider has unsupported on-chain status") + registrar := newFakeRegistrar(func(int) ([]byte, []byte, uint64, error) { + return nil, nil, 0, failure + }) + cfg := DefaultReconcilerConfig() + cfg.MaxAttempts = 3 + cfg.BaseBackoff = time.Millisecond // keep the test fast + reconciler := NewReconciler(store, store, registrar, cfg) + + for i := 0; i < 3; i++ { + reconciler.ReconcileOnce(context.Background()) + // Force the next row due regardless of backoff so each loop + // iteration actually attempts again (unit test, not real time). + store.rows["provider-1"].nextAttempt = time.Time{} + } + + row := store.rows["provider-1"] + if row.state != "FAILED" { + t.Fatalf("expected FAILED after exceeding MaxAttempts, got %s (attempts=%d)", row.state, row.attemptCount) + } + if row.attemptCount != 3 { + t.Fatalf("expected 3 recorded attempts, got %d", row.attemptCount) + } + // A FAILED row must stop being picked up -- it is a terminal, explicit + // failure, not a silent infinite retry loop. + due, err := store.DueChainRegistrations(context.Background(), 10) + if err != nil { + t.Fatalf("DueChainRegistrations: %v", err) + } + if len(due) != 0 { + t.Fatalf("expected a FAILED registration to no longer be due, got %+v", due) + } +} + +func TestReconcileRejectsMalformedPublicKeyWithoutCallingTheRegistrar(t *testing.T) { + store := newFakeChainStore() + store.addReady("provider-1", []byte("too-short")) + registrar := newFakeRegistrar(func(int) ([]byte, []byte, uint64, error) { + t.Fatal("registrar must not be called for a malformed stored key") + return nil, nil, 0, nil + }) + reconciler := NewReconciler(store, store, registrar, DefaultReconcilerConfig()) + + reconciler.ReconcileOnce(context.Background()) + + row := store.rows["provider-1"] + if row.state != "FAILED" { + t.Fatalf("expected an immediate terminal FAILED for malformed data, got %s", row.state) + } +} + +func TestReconcileOnceProcessesAWholeDueBatch(t *testing.T) { + store := newFakeChainStore() + store.addReady("provider-1", testPublicKey(1)) + store.addReady("provider-2", testPublicKey(2)) + store.addReady("provider-3", testPublicKey(3)) + registrar := newFakeRegistrar(func(int) ([]byte, []byte, uint64, error) { + return []byte("extrinsic"), []byte("block"), 1, nil + }) + reconciler := NewReconciler(store, store, registrar, DefaultReconcilerConfig()) + + reconciler.ReconcileOnce(context.Background()) + + for _, providerID := range []string{"provider-1", "provider-2", "provider-3"} { + if store.rows[providerID].state != "FINALIZED" { + t.Fatalf("expected %s to be FINALIZED, got %s", providerID, store.rows[providerID].state) + } + } +} + +func TestReconcilerSurvivesRestartByReadingAttemptCountFromTheStore(t *testing.T) { + // Simulates a Control Plane restart: attempt state lives only in the + // store (as it would in Postgres), never in the Reconciler's memory, so + // a brand new Reconciler instance continues backoff/attempt counting + // correctly instead of resetting it. + store := newFakeChainStore() + store.addReady("provider-1", testPublicKey(1)) + failure := errors.New("chain unavailable") + registrar := newFakeRegistrar(func(int) ([]byte, []byte, uint64, error) { + return nil, nil, 0, failure + }) + cfg := DefaultReconcilerConfig() + cfg.BaseBackoff = time.Millisecond + + first := NewReconciler(store, store, registrar, cfg) + first.ReconcileOnce(context.Background()) + if store.rows["provider-1"].attemptCount != 1 { + t.Fatalf("expected attempt_count 1 after first reconciler's pass") + } + + store.rows["provider-1"].nextAttempt = time.Time{} // force due again + second := NewReconciler(store, store, registrar, cfg) + second.ReconcileOnce(context.Background()) + if store.rows["provider-1"].attemptCount != 2 { + t.Fatalf("expected attempt_count 2 after the 'restarted' reconciler's pass, got %d", store.rows["provider-1"].attemptCount) + } +} + +func TestReconcileOnceIsIdempotentAgainstDuplicateDelivery(t *testing.T) { + // A FINALIZED row must never be reprocessed, whether the duplicate + // trigger is a second reconcile pass or a concurrent Agent CompleteJoin + // retry racing the Reconciler for the same provider. + store := newFakeChainStore() + store.addReady("provider-1", testPublicKey(1)) + activations := 0 + registrar := newFakeRegistrar(func(int) ([]byte, []byte, uint64, error) { + return []byte("extrinsic"), []byte("block"), 7, nil + }) + reconciler := NewReconciler(store, activationCountingStore{store, &activations}, registrar, DefaultReconcilerConfig()) + + reconciler.ReconcileOnce(context.Background()) + reconciler.ReconcileOnce(context.Background()) + reconciler.ReconcileOnce(context.Background()) + + if activations != 1 { + t.Fatalf("expected exactly one activation despite repeated reconcile passes, got %d", activations) + } + if registrar.calls[string(testPublicKey(1))] != 1 { + t.Fatalf("expected the registrar to be called exactly once once FINALIZED, got %d calls", registrar.calls[string(testPublicKey(1))]) + } +} + +// activationCountingStore wraps fakeChainStore's Activator to count calls +// without changing fakeChainStore's own behavior/assertions elsewhere. +type activationCountingStore struct { + *fakeChainStore + count *int +} + +func (s activationCountingStore) ActivateProvider(ctx context.Context, providerID string, finalization ChainFinalization) (Completion, error) { + *s.count++ + return s.fakeChainStore.ActivateProvider(ctx, providerID, finalization) +} + +func TestReconcileOnceIgnoresRegistrationsNotYetDue(t *testing.T) { + store := newFakeChainStore() + store.addReady("provider-1", testPublicKey(1)) + store.rows["provider-1"].state = "RETRY" + store.rows["provider-1"].nextAttempt = time.Now().Add(time.Hour) + registrar := newFakeRegistrar(func(int) ([]byte, []byte, uint64, error) { + t.Fatal("registrar must not be called before next_attempt_at") + return nil, nil, 0, nil + }) + reconciler := NewReconciler(store, store, registrar, DefaultReconcilerConfig()) + + reconciler.ReconcileOnce(context.Background()) +} + +func TestBackoffForDoublesAndCapsAtMaxBackoff(t *testing.T) { + cfg := ReconcilerConfig{BaseBackoff: time.Second, MaxBackoff: 10 * time.Second} + reconciler := NewReconciler(newFakeChainStore(), newFakeChainStore(), nil, cfg) + + cases := []struct { + attempt int + want time.Duration + }{ + {1, time.Second}, + {2, 2 * time.Second}, + {3, 4 * time.Second}, + {4, 8 * time.Second}, + {5, 10 * time.Second}, // would be 16s uncapped + {100, 10 * time.Second}, + } + for _, tc := range cases { + if got := reconciler.backoffFor(tc.attempt); got != tc.want { + t.Errorf("backoffFor(%d) = %v, want %v", tc.attempt, got, tc.want) + } + } +} + +func TestRunStopsWhenContextIsCancelled(t *testing.T) { + store := newFakeChainStore() + registrar := newFakeRegistrar(func(int) ([]byte, []byte, uint64, error) { return nil, nil, 0, nil }) + cfg := DefaultReconcilerConfig() + cfg.Interval = time.Millisecond + reconciler := NewReconciler(store, store, registrar, cfg) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + reconciler.Run(ctx) + close(done) + }() + cancel() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("Run did not return after context cancellation") + } +}