diff --git a/cmd/cli/main.go b/cmd/cli/main.go index 25fc1ec..03fb785 100644 --- a/cmd/cli/main.go +++ b/cmd/cli/main.go @@ -34,6 +34,7 @@ import ( "log/slog" "os" "slices" + "strconv" "strings" "time" @@ -80,6 +81,8 @@ func main() { runAddModels() case "remove-models": runRemoveModels() + case "abandon-stuck": + runAbandonStuck() case "reinstate": runReinstate() case "top-up-stake": @@ -125,6 +128,14 @@ Commands: keygen Generate/load ECDH encryption key and print public key hex register Register worker on-chain (stake + encryption key + models) add-models Add models to an already-registered worker + remove-models + Remove the models named in SUPPORTED_MODELS from the worker's + on-chain list (sortition routes by the registry, not the env) + abandon-stuck + Hand back every job of this worker still Submitted/Acknowledged + on chain (JobRegistry.abandonJob): refunds the consumer, frees the + in-flight slot. ABANDON_LOOKBACK_BLOCKS (default 150000) bounds + the scan, ABANDON_DRY_RUN=1 only lists. drain Mark worker ineligible for new sessions (selection-only) undrain Reverse drain — restore worker eligibility deregister Deregister worker and withdraw stake @@ -347,6 +358,43 @@ func runRemoveModels() { } } +// runAbandonStuck sweeps this worker's in-flight backlog (see cli.AbandonStuck). +// Each abandon is one transaction, so the budget is txTimeout per job plus the scan. +func runAbandonStuck() { + cfg, logger := loadAndValidateForSettlement() + signingKey, workerAddr := loadSigningKey(cfg, logger) + + lookback := uint64(150_000) + if v := os.Getenv("ABANDON_LOOKBACK_BLOCKS"); v != "" { + n, err := strconv.ParseUint(v, 10, 64) + if err != nil { + logger.Error("ABANDON_LOOKBACK_BLOCKS must be a whole number of blocks", "value", v) + os.Exit(2) + } + lookback = n + } + dryRun := os.Getenv("ABANDON_DRY_RUN") == "1" + + chainClient := dialChain(cfg, signingKey, logger) + defer chainClient.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute) + defer cancel() + + n, err := cli.AbandonStuck(ctx, cli.AbandonStuckOpts{ + Client: chainClient, + Worker: workerAddr, + LookbackBlocks: lookback, + DryRun: dryRun, + Out: os.Stdout, + Logger: logger, + }) + if err != nil { + logger.Error("abandon-stuck finished with errors", "abandoned", n, "error", err) + os.Exit(1) + } +} + func runDeregister() { cfg, logger := loadAndValidateCfg() signingKey, workerAddr := loadSigningKey(cfg, logger) diff --git a/internal/cli/abandon_stuck.go b/internal/cli/abandon_stuck.go new file mode 100644 index 0000000..f58e08a --- /dev/null +++ b/internal/cli/abandon_stuck.go @@ -0,0 +1,129 @@ +package cli + +import ( + "context" + "fmt" + "io" + "log/slog" + + "github.com/ethereum/go-ethereum/common" + + "github.com/lightchain/worker/internal/chain" +) + +// StuckChain is the narrow chain surface the abandon-stuck command needs. +type StuckChain interface { + Head(ctx context.Context) (chain.HeadInfo, error) + FilterJobSubmitted(ctx context.Context, fromBlock, toBlock uint64) ([]chain.JobSubmittedEvent, error) + GetJobState(ctx context.Context, jobID uint64) (chain.JobStateInfo, error) + AbandonJob(ctx context.Context, jobID uint64) error +} + +// AbandonStuckOpts configures one backlog sweep. +type AbandonStuckOpts struct { + Client StuckChain + Worker common.Address + LookbackBlocks uint64 // how far behind the head to look for this worker's jobs + ChunkSize uint64 // eth_getLogs range per request (RPC limits), default 5000 + DryRun bool // list, do not send + Out io.Writer + Logger *slog.Logger +} + +// AbandonStuck hands back every job of this worker that is still Submitted or +// Acknowledged on chain (JobRegistry.abandonJob): the consumer is refunded and +// the job leaves the worker's in-flight count at once. +// +// Why a separate command: the worker abandons a job it fails from now on, but a +// backlog from before that change stays Acknowledged until the keeper's +// no-fault settle 48 h after each deadline. At maxConcurrentJobs of those the +// chain refuses the worker's session claims (2026-10-19, g1 worker-4/5), and +// WorkerRegistry.removeSupportedModel refuses while any job for the model is +// still counted active, so the backlog blocks both recovery paths. +// +// Jobs are found through JobSubmitted events in the look-back window; anything +// older has been swept by the keeper already. Each candidate's state is read +// before the transaction, and a failed abandon is reported and skipped, never +// fatal, so one odd job cannot stop the sweep. +func AbandonStuck(ctx context.Context, o AbandonStuckOpts) (abandoned int, err error) { + if o.Client == nil { + return 0, fmt.Errorf("abandon-stuck: no chain client") + } + chunk := o.ChunkSize + if chunk == 0 { + chunk = 5000 + } + log := o.Logger + if log == nil { + log = slog.Default() + } + + head, err := o.Client.Head(ctx) + if err != nil { + return 0, fmt.Errorf("abandon-stuck: head: %w", err) + } + from := uint64(1) + if head.Number > o.LookbackBlocks { + from = head.Number - o.LookbackBlocks + } + + // Phase 1: this worker's jobs in the window. + var mine []uint64 + for lo := from; lo <= head.Number; lo += chunk { + hi := lo + chunk - 1 + if hi > head.Number { + hi = head.Number + } + evs, err := o.Client.FilterJobSubmitted(ctx, lo, hi) + if err != nil { + return 0, fmt.Errorf("abandon-stuck: JobSubmitted %d-%d: %w", lo, hi, err) + } + for _, ev := range evs { + if ev.Worker == o.Worker { + mine = append(mine, ev.JobID) + } + } + } + _, _ = fmt.Fprintf(o.Out, "worker %s: %d jobs submitted in blocks %d-%d\n", o.Worker.Hex(), len(mine), from, head.Number) + + // Phase 2: the ones still in flight. + var stuck []uint64 + for _, id := range mine { + st, err := o.Client.GetJobState(ctx, id) + if err != nil { + log.Warn("getJob failed; skipping", "jobID", id, "error", err) + continue + } + if st.State == chain.JobStateSubmitted || st.State == chain.JobStateAcknowledged { + stuck = append(stuck, id) + } + } + _, _ = fmt.Fprintf(o.Out, "in flight (Submitted/Acknowledged): %d\n", len(stuck)) + if len(stuck) == 0 { + return 0, nil + } + if o.DryRun { + _, _ = fmt.Fprintf(o.Out, "dry run: would abandon %v\n", stuck) + return 0, nil + } + + // Phase 3: hand them back, one tx each. + var failed []uint64 + for _, id := range stuck { + if err := ctx.Err(); err != nil { + return abandoned, fmt.Errorf("abandon-stuck: stopped after %d of %d: %w", abandoned, len(stuck), err) + } + if err := o.Client.AbandonJob(ctx, id); err != nil { + log.Warn("abandonJob failed; skipping", "jobID", id, "error", err) + failed = append(failed, id) + continue + } + abandoned++ + log.Info("job abandoned: fee refunded, in-flight slot released", "jobID", id) + } + _, _ = fmt.Fprintf(o.Out, "abandoned %d of %d\n", abandoned, len(stuck)) + if len(failed) > 0 { + return abandoned, fmt.Errorf("%d job(s) could not be abandoned (already settled, or the tx reverted): %v", len(failed), failed) + } + return abandoned, nil +} diff --git a/internal/cli/abandon_stuck_test.go b/internal/cli/abandon_stuck_test.go new file mode 100644 index 0000000..57cd06f --- /dev/null +++ b/internal/cli/abandon_stuck_test.go @@ -0,0 +1,110 @@ +package cli + +import ( + "bytes" + "context" + "errors" + "testing" + + "github.com/ethereum/go-ethereum/common" + "github.com/stretchr/testify/require" + + "github.com/lightchain/worker/internal/chain" +) + +type stuckChain struct { + head uint64 + events []chain.JobSubmittedEvent + states map[uint64]chain.JobStateInfo + abandoned []uint64 + failOn uint64 + ranges [][2]uint64 +} + +func (s *stuckChain) Head(context.Context) (chain.HeadInfo, error) { + return chain.HeadInfo{Number: s.head}, nil +} + +func (s *stuckChain) FilterJobSubmitted(_ context.Context, from, to uint64) ([]chain.JobSubmittedEvent, error) { + s.ranges = append(s.ranges, [2]uint64{from, to}) + var out []chain.JobSubmittedEvent + for _, ev := range s.events { + if ev.BlockNumber >= from && ev.BlockNumber <= to { + out = append(out, ev) + } + } + return out, nil +} + +func (s *stuckChain) GetJobState(_ context.Context, id uint64) (chain.JobStateInfo, error) { + st, ok := s.states[id] + if !ok { + return chain.JobStateInfo{}, errors.New("no such job") + } + return st, nil +} + +func (s *stuckChain) AbandonJob(_ context.Context, id uint64) error { + if id == s.failOn { + return errors.New("execution reverted") + } + s.abandoned = append(s.abandoned, id) + return nil +} + +var ( + me = common.HexToAddress("0x00000000000000000000000000000000000000a1") + other = common.HexToAddress("0x00000000000000000000000000000000000000b2") +) + +func newStuckChain() *stuckChain { + return &stuckChain{ + head: 10_000, + events: []chain.JobSubmittedEvent{ + {JobID: 1, Worker: me, BlockNumber: 100}, // outside the look-back + {JobID: 2, Worker: me, BlockNumber: 6_000}, // acknowledged, mine + {JobID: 3, Worker: other, BlockNumber: 6_100}, + {JobID: 4, Worker: me, BlockNumber: 7_000}, // completed + {JobID: 5, Worker: me, BlockNumber: 9_900}, // submitted, mine + }, + states: map[uint64]chain.JobStateInfo{ + 1: {State: chain.JobStateAcknowledged, Worker: me}, + 2: {State: chain.JobStateAcknowledged, Worker: me}, + 3: {State: chain.JobStateAcknowledged, Worker: other}, + 4: {State: chain.JobStateCompleted, Worker: me}, + 5: {State: chain.JobStateSubmitted, Worker: me}, + }, + } +} + +func TestAbandonStuck_AbandonsOnlyMyInFlightJobsInWindow(t *testing.T) { + c := newStuckChain() + var out bytes.Buffer + n, err := AbandonStuck(context.Background(), AbandonStuckOpts{Client: c, Worker: me, LookbackBlocks: 5_000, ChunkSize: 2_000, Out: &out}) + require.NoError(t, err) + require.Equal(t, 2, n) + require.Equal(t, []uint64{2, 5}, c.abandoned, "job 1 is outside the window, 3 belongs to another worker, 4 is completed") + require.Equal(t, [2]uint64{5_000, 6_999}, c.ranges[0], "scan starts head - lookback and respects the chunk size") + require.Contains(t, out.String(), "abandoned 2 of 2") +} + +func TestAbandonStuck_DryRunSendsNothing(t *testing.T) { + c := newStuckChain() + var out bytes.Buffer + n, err := AbandonStuck(context.Background(), AbandonStuckOpts{Client: c, Worker: me, LookbackBlocks: 5_000, DryRun: true, Out: &out}) + require.NoError(t, err) + require.Zero(t, n) + require.Empty(t, c.abandoned) + require.Contains(t, out.String(), "would abandon [2 5]") +} + +func TestAbandonStuck_OneRevertDoesNotStopTheSweep(t *testing.T) { + c := newStuckChain() + c.failOn = 2 + var out bytes.Buffer + n, err := AbandonStuck(context.Background(), AbandonStuckOpts{Client: c, Worker: me, LookbackBlocks: 5_000, Out: &out}) + require.Error(t, err) + require.Equal(t, 1, n) + require.Equal(t, []uint64{5}, c.abandoned) + require.Contains(t, err.Error(), "[2]") +}