feat: replace task scheduler with hydra

[Dodecahedr0x]

Closes #1375

Summary

First step toward moving toward an external crank service based on Hydra. This PR preserves backward compatibility by forward legacy crank requests to the Hydra program, with the validator being the sponsor of those cranks. Any cranks currently in the SQLite DB will be migrated

Breaking Changes

• Yes

The task scheduler config is now removed, so validators that were specifying values need to remove them.

Important

This shouldn't be merged until the ephemeral hydra program is deployed and validators are provided with a faucet keypair (will be delegated, so one per validator) funded with enough SOL to prepay for current and future cranks

Also, it requires running the following script to convert the existing DB to hydra crank

Tested using the following script:

python migration-script.py --storage-path path --rpc-url url --identity-keypair keypair --faucet-keypair keypair

"""One-time, out-of-band migration of tasks persisted by the legacy
(validator-funded) task scheduler onto the hydra crank program.

Run this against a running validator (RPC reachable, faucet delegated and
funded) before removing the legacy `task_scheduler.sqlite` database. The
runtime task scheduler no longer performs this migration itself.

Pure stdlib + openssl (for Ed25519 signing) — no third-party dependencies.
"""
import argparse
import base64
import hashlib
import json
import os
import subprocess
import sqlite3
import sys
import tempfile
import time
import urllib.error
import urllib.request

# --- base58 --------------------------------------------------------------

B58_ALPHABET = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz"


def b58encode(b: bytes) -> str:
    n = int.from_bytes(b, "big")
    leading_zeros = len(b) - len(b.lstrip(b"\x00"))
    if n == 0:
        return "1" * len(b)
    digits = []
    while n > 0:
        n, r = divmod(n, 58)
        digits.append(B58_ALPHABET[r])
    return "1" * leading_zeros + "".join(reversed(digits))


def b58decode(s: str, length: int) -> bytes:
    n = 0
    for c in s:
        n = n * 58 + B58_ALPHABET.index(c)
    return n.to_bytes(length, "big")


# --- ed25519 curve check, for find_program_address ------------------------

_P = 2**255 - 19
_D = (-121665 * pow(121666, _P - 2, _P)) % _P


def _is_on_curve(compressed: bytes) -> bool:
    y = int.from_bytes(compressed, "little") & ((1 << 255) - 1)
    if y >= _P:
        return True  # invalid encoding -> reject this candidate
    yy = (y * y) % _P
    u = (yy - 1) % _P
    v = (_D * yy + 1) % _P
    w = (u * v) % _P
    if w == 0:
        return True
    return pow(w, (_P - 1) // 2, _P) == 1


PDA_MARKER = b"ProgramDerivedAddress"


def find_program_address(seeds: list, program_id: bytes):
    for bump in range(255, -1, -1):
        h = hashlib.sha256()
        for s in seeds:
            h.update(s)
        h.update(bytes([bump]))
        h.update(program_id)
        h.update(PDA_MARKER)
        candidate = h.digest()
        if not _is_on_curve(candidate):
            return candidate, bump
    raise ValueError("no viable bump seed found")


# --- compact-u16 ("short vec") + legacy transaction assembly --------------

def compact_u16(n: int) -> bytes:
    out = bytearray()
    while True:
        b = n & 0x7F
        n >>= 7
        if n == 0:
            out.append(b)
            break
        out.append(b | 0x80)
    return bytes(out)


def serialize_transaction(signatures: list, message: bytes) -> bytes:
    return compact_u16(len(signatures)) + b"".join(signatures) + message


class TxBuilder:
    """Builds a legacy (non-versioned) Solana Message + Transaction.

    The leading accounts are always-writable signers with the fee payer first.
    Extra accounts referenced by instructions are deduped and bucketed
    (writable-then-readonly) after the signers.
    """

    def __init__(self, signers: list):
        self.accounts = list(signers)
        self.writable = [True] * len(signers)
        self.num_signers = len(signers)
        self.instructions = []

    def _account_index(self, pubkey: bytes, writable: bool) -> int:
        for i, pk in enumerate(self.accounts):
            if pk == pubkey:
                if writable and not self.writable[i]:
                    self.writable[i] = True
                return i
        self.accounts.append(pubkey)
        self.writable.append(writable)
        return len(self.accounts) - 1

    def add_instruction(self, program_id: bytes, accounts: list, data: bytes):
        prog_idx = self._account_index(program_id, False)
        idxs = [self._account_index(pk, w) for pk, w in accounts]
        self.instructions.append((prog_idx, idxs, data))

    def compile_message(self, blockhash: bytes) -> bytes:
        n = len(self.accounts)
        writable_extra = [i for i in range(self.num_signers, n) if self.writable[i]]
        readonly_extra = [i for i in range(self.num_signers, n) if not self.writable[i]]
        new_order = list(range(self.num_signers)) + writable_extra + readonly_extra
        old_to_new = {old: new for new, old in enumerate(new_order)}

        header = bytes([self.num_signers, 0, len(readonly_extra)])
        out = bytearray()
        out += header
        out += compact_u16(len(new_order))
        out += b"".join(self.accounts[i] for i in new_order)
        out += blockhash
        out += compact_u16(len(self.instructions))
        for prog_idx, idxs, data in self.instructions:
            out.append(old_to_new[prog_idx])
            remapped = [old_to_new[i] for i in idxs]
            out += compact_u16(len(remapped))
            out += bytes(remapped)
            out += compact_u16(len(data))
            out += data
        return bytes(out)


# --- bincode decode of Vec<Instruction>, as stored by the legacy scheduler's
# sqlite db (bincode::serialize(&Vec<solana_instruction::Instruction>)) ------

class Instr:
    __slots__ = ("program_id", "accounts", "data")  # accounts: [(pubkey, is_signer, is_writable)]

    def __init__(self, program_id, accounts, data):
        self.program_id = program_id
        self.accounts = accounts
        self.data = data


def decode_bincode_instructions(blob: bytes):
    pos = 0

    def read_u64():
        nonlocal pos
        v = int.from_bytes(blob[pos:pos + 8], "little")
        pos += 8
        return v

    def read_bytes(n):
        nonlocal pos
        v = blob[pos:pos + n]
        pos += n
        return v

    count = read_u64()
    out = []
    for _ in range(count):
        program_id = read_bytes(32)
        n_accounts = read_u64()
        accounts = []
        for _ in range(n_accounts):
            pk = read_bytes(32)
            is_signer = read_bytes(1) != b"\x00"
            is_writable = read_bytes(1) != b"\x00"
            accounts.append((pk, is_signer, is_writable))
        data_len = read_u64()
        data = read_bytes(data_len)
        out.append(Instr(program_id, accounts, data))
    if pos != len(blob):
        raise ValueError(f"trailing bytes after decoding instructions: {len(blob) - pos}")
    return out


# --- hydra ephemeral-program instructions ---------------------------------
# Wire format + program/account IDs pinned to hydra-api rev
# 449532ee0ee792e9a3809b16fd9d289c2024341a (see Cargo.toml).

HYDRA_EPHEMERAL_PROGRAM_ID = b58decode("eHyd5BU8QffvHi4GnXwxrK4WpS7pM2x9UGKHBWii7mf", 32)
MAGIC_VAULT_ID = b58decode("MagicVau1t999999999999999999999999999999999", 32)
MAGIC_PROGRAM_ID = b58decode("Magic11111111111111111111111111111111111111", 32)
SYSTEM_PROGRAM_ID = bytes(32)

HYDRA_IX_CREATE = 0
HYDRA_IX_CANCEL = 2

CRANK_SEED_PREFIX = b"crank"
CRANKER_REWARD = 10_000  # 2 * BASE_FEE_LAMPORTS(5000) (non-"ephemeral"-feature build)
CRANK_HEADER_SIZE = 120
SERIALIZED_META_SIZE = 33
ACCOUNT_STATIC_SIZE = 60  # solana_account::AccountSharedData::ACCOUNT_STATIC_SIZE (pinned rev)
EPHEMERAL_RENT_PER_BYTE = 32
META_FLAG_WRITABLE = 0b10


def crank_pubkey(authority: bytes, task_id: int):
    seed_hash = hashlib.sha256(authority + task_id.to_bytes(8, "little", signed=True)).digest()
    return find_program_address([CRANK_SEED_PREFIX, seed_hash], HYDRA_EPHEMERAL_PROGRAM_ID)


def crank_rent_floor(instructions) -> int:
    region_len = CRANK_HEADER_SIZE
    for ix in instructions:
        region_len += 2 + len(ix.accounts) * SERIALIZED_META_SIZE + 32 + 2 + len(ix.data)
    total_size = region_len + ACCOUNT_STATIC_SIZE
    return total_size * EPHEMERAL_RENT_PER_BYTE


def build_create_ix(faucet: bytes, authority: bytes, task_id: int, crank: bytes,
                     start_slot: int, interval_slots: int, iterations: int, instructions):
    seed_hash = hashlib.sha256(authority + task_id.to_bytes(8, "little", signed=True)).digest()
    data = bytearray()
    data.append(HYDRA_IX_CREATE)
    data += seed_hash
    data += faucet  # CreateArgs.authority = faucet pubkey (cancel authority)
    data += start_slot.to_bytes(8, "little")
    data += interval_slots.to_bytes(8, "little")
    data += iterations.to_bytes(8, "little")
    data += (0).to_bytes(8, "little")  # priority_tip
    data += (0).to_bytes(4, "little")  # cu_limit
    for ix in instructions:
        data.append(len(ix.accounts))
        data += len(ix.data).to_bytes(2, "little")
        data += ix.program_id
        for pk, _is_signer, is_writable in ix.accounts:
            data.append(META_FLAG_WRITABLE if is_writable else 0)
            data += pk
        data += ix.data
    accounts = [
        (faucet, True),
        (crank, True),
        (MAGIC_VAULT_ID, True),
        (MAGIC_PROGRAM_ID, False),
    ]
    return HYDRA_EPHEMERAL_PROGRAM_ID, accounts, bytes(data)


def build_cancel_ix(faucet: bytes, crank: bytes, recipient: bytes):
    accounts = [
        (faucet, True),
        (crank, True),
        (recipient, True),
        (MAGIC_VAULT_ID, True),
        (MAGIC_PROGRAM_ID, False),
    ]
    return HYDRA_EPHEMERAL_PROGRAM_ID, accounts, bytes([HYDRA_IX_CANCEL])


def transfer_ix(frm: bytes, to: bytes, lamports: int):
    data = (2).to_bytes(4, "little") + lamports.to_bytes(8, "little")
    return SYSTEM_PROGRAM_ID, [(frm, True), (to, True)], data


def is_valid_task_interval(interval_millis: int) -> bool:
    return 0 < interval_millis < (2**32 - 1)


def interval_slots(interval_millis: int, slot_interval_millis: int) -> int:
    slot_millis = max(slot_interval_millis, 1)
    interval_millis = max(interval_millis, 0)
    slots = (interval_millis + slot_millis - 1) // slot_millis
    return max(slots, 1)


def legacy_start_slot(last_execution_millis: int, interval_millis: int, current_millis: int,
                       current_slot: int, slot_interval_millis: int) -> int:
    if last_execution_millis <= 0:
        return current_slot
    next_execution_millis = last_execution_millis + max(interval_millis, 0)
    remaining_millis = max(next_execution_millis - current_millis, 0)
    if remaining_millis == 0:
        return current_slot
    return current_slot + interval_slots(remaining_millis, slot_interval_millis)


# --- RPC client + keypair/signing helpers ---------------------------------

class RpcError(Exception):
    pass


class RpcClient:
    def __init__(self, url: str):
        self.url = url
        self._id = 0

    def call(self, method: str, params: list):
        self._id += 1
        payload = json.dumps({"jsonrpc": "2.0", "id": self._id, "method": method, "params": params}).encode()
        req = urllib.request.Request(self.url, data=payload, headers={"Content-Type": "application/json"})
        try:
            with urllib.request.urlopen(req, timeout=15) as resp:
                body = json.loads(resp.read())
        except urllib.error.URLError as e:
            raise RpcError(f"{method} request failed: {e}") from e
        if "error" in body:
            raise RpcError(f"{method} rpc error: {body['error']}")
        return body["result"]

    def get_latest_blockhash(self) -> bytes:
        result = self.call("getLatestBlockhash", [{"commitment": "confirmed"}])
        return b58decode(result["value"]["blockhash"], 32)

    def get_slot(self) -> int:
        return self.call("getSlot", [{"commitment": "confirmed"}])

    def get_account_owner_and_lamports(self, pubkey: bytes):
        """(owner_bytes_or_None, lamports); owner is None if the account doesn't exist."""
        result = self.call(
            "getAccountInfo",
            [b58encode(pubkey), {"encoding": "base64", "commitment": "confirmed"}],
        )
        value = result["value"]
        if value is None:
            return None, 0
        return b58decode(value["owner"], 32), value["lamports"]

    def send_transaction(self, raw_tx: bytes) -> str:
        return self.call(
            "sendTransaction",
            [base64.b64encode(raw_tx).decode(),
             {"encoding": "base64", "preflightCommitment": "confirmed", "skipPreflight": True}],
        )

    def confirm_signature(self, sig: str, timeout_s: float = 30.0) -> bool:
        deadline = time.monotonic() + timeout_s
        while time.monotonic() < deadline:
            result = self.call("getSignatureStatuses", [[sig], {"searchTransactionHistory": True}])
            status = result["value"][0]
            if status is not None:
                if status.get("err") is not None:
                    raise RpcError(f"transaction {sig} failed: {status['err']}")
                if status.get("confirmationStatus") in ("confirmed", "finalized"):
                    return True
            time.sleep(0.3)
        raise RpcError(f"timed out waiting for confirmation of {sig}")


def read_solana_keypair_file(path: str):
    """Returns (seed32, pubkey32) from a standard solana-keygen JSON keypair file."""
    with open(path) as f:
        arr = json.load(f)
    if len(arr) != 64:
        raise ValueError(f"{path}: expected a 64-byte Solana keypair array, got {len(arr)}")
    raw = bytes(arr)
    return raw[:32], raw[32:]


def ed25519_sign(seed32: bytes, message: bytes) -> bytes:
    # Wrap the raw 32-byte seed in the fixed PKCS8 DER prefix for Ed25519
    # (RFC 8410) so openssl can load it as a private key.
    der = bytes.fromhex("302e020100300506032b657004220420") + seed32
    with tempfile.NamedTemporaryFile(delete=False) as kf:
        kf.write(der)
        keypath = kf.name
    with tempfile.NamedTemporaryFile(delete=False) as mf:
        mf.write(message)
        msgpath = mf.name
    try:
        p = subprocess.run(
            ["openssl", "pkeyutl", "-sign", "-rawin", "-inkey", keypath, "-keyform", "DER", "-in", msgpath],
            capture_output=True,
        )
        if p.returncode != 0:
            raise RuntimeError(f"openssl ed25519 sign failed: {p.stderr.decode()}")
        return p.stdout
    finally:
        os.unlink(keypath)
        os.unlink(msgpath)


# --- migration orchestration -----------------------------------------------

BLOCK_READY_TIMEOUT_S = 60
FAUCET_READY_TIMEOUT_S = 60


def log(msg):
    print(f"[migrate-task-scheduler] {msg}", file=sys.stderr, flush=True)


def wait_for_block_ready(rpc: RpcClient):
    deadline = time.monotonic() + BLOCK_READY_TIMEOUT_S
    while time.monotonic() < deadline:
        try:
            rpc.get_latest_blockhash()
            return
        except RpcError:
            time.sleep(0.2)
    raise RpcError("timed out waiting for a usable blockhash before migration")


def wait_for_faucet_ready(rpc: RpcClient, faucet_pubkey: bytes):
    deadline = time.monotonic() + FAUCET_READY_TIMEOUT_S
    while time.monotonic() < deadline:
        try:
            _owner, lamports = rpc.get_account_owner_and_lamports(faucet_pubkey)
            if lamports > 0:
                return
        except RpcError:
            pass
        time.sleep(0.2)
    raise RpcError(
        f"timed out waiting for faucet {b58encode(faucet_pubkey)} to be delegated before paying cranks"
    )


def migrate(db_path, rpc_url, faucet_seed, faucet_pub, block_time_ms):
    rpc = RpcClient(rpc_url)
    conn = sqlite3.connect(db_path)
    conn.execute(
        """CREATE TABLE IF NOT EXISTS tasks (
            id INTEGER PRIMARY KEY,
            instructions BLOB NOT NULL,
            authority TEXT NOT NULL,
            execution_interval_millis INTEGER NOT NULL,
            executions_left INTEGER NOT NULL,
            last_execution_millis INTEGER NOT NULL DEFAULT 0,
            created_at INTEGER NOT NULL DEFAULT 0,
            updated_at INTEGER NOT NULL DEFAULT 0
        )"""
    )
    rows = conn.execute(
        "SELECT id, instructions, authority, execution_interval_millis, executions_left, last_execution_millis "
        "FROM tasks"
    ).fetchall()

    if not rows:
        log("No persisted tasks to migrate")
        return

    log(f"Migrating {len(rows)} persisted task(s) onto hydra")

    valid = []
    for row in rows:
        task_id, _blob, _auth, interval_millis, executions_left, _last = row
        if is_valid_task_interval(interval_millis) and executions_left > 0:
            valid.append(row)
        else:
            log(
                f"Dropping invalid task {task_id} during migration "
                f"(interval={interval_millis}, executions_left={executions_left})"
            )
            conn.execute("DELETE FROM tasks WHERE id = ?", (task_id,))
            conn.commit()

    if not valid:
        log("No valid tasks left to migrate")
        return

    wait_for_block_ready(rpc)
    log("Migration: block ready")
    wait_for_faucet_ready(rpc, faucet_pub)
    log("Migration: faucet ready")

    for row in valid:
        task_id, instructions_blob, authority_str, interval_millis, executions_left, last_execution_millis = row
        log(f"Migration: creating crank for task {task_id}")
        try:
            authority = b58decode(authority_str, 32)
            instructions = decode_bincode_instructions(instructions_blob)
            crank, _bump = crank_pubkey(authority, task_id)

            owner, _lamports = rpc.get_account_owner_and_lamports(crank)
            crank_exists = owner == HYDRA_EPHEMERAL_PROGRAM_ID

            blockhash = rpc.get_latest_blockhash()
            slot = rpc.get_slot()
            now_millis = int(time.time() * 1000)
            start_slot = legacy_start_slot(last_execution_millis, interval_millis, now_millis, slot, block_time_ms)
            iv_slots = interval_slots(interval_millis, block_time_ms)

            create_prog, create_accs, create_data = build_create_ix(
                faucet_pub, authority, task_id, crank, start_slot, iv_slots, executions_left, instructions
            )
            funding = executions_left * CRANKER_REWARD + crank_rent_floor(instructions)
            fund_prog, fund_accs, fund_data = transfer_ix(faucet_pub, crank, funding)

            tb = TxBuilder([faucet_pub])
            if crank_exists:
                cancel_prog, cancel_accs, cancel_data = build_cancel_ix(faucet_pub, crank, faucet_pub)
                tb.add_instruction(cancel_prog, cancel_accs, cancel_data)
            tb.add_instruction(create_prog, create_accs, create_data)
            tb.add_instruction(fund_prog, fund_accs, fund_data)

            message = tb.compile_message(blockhash)
            sig = ed25519_sign(faucet_seed, message)
            raw_tx = serialize_transaction([sig], message)

            sig_b58 = rpc.send_transaction(raw_tx)
            rpc.confirm_signature(sig_b58)

            conn.execute("DELETE FROM tasks WHERE id = ?", (task_id,))
            conn.commit()
            log(f"Migration: created crank for task {task_id} (tx {sig_b58})")
        except Exception as e:
            log(f"Failed to migrate task {task_id} onto hydra: {e}")

    log("Task migration complete")


def main():
    parser = argparse.ArgumentParser(prog="migrate-task-scheduler.sh", description=__doc__)
    parser.add_argument("--storage-path", required=True,
                         help="Root storage directory the validator was started with; the legacy "
                              "database lives at <storage-path>/task_scheduler.sqlite")
    parser.add_argument("--rpc-url", required=True, help="RPC URL of the running validator")
    parser.add_argument("--faucet-keypair", required=True,
                         help="Path to the task scheduler's faucet keypair JSON file")
    parser.add_argument("--block-time-ms", type=int, default=50, help="Validator block time in milliseconds")
    args = parser.parse_args()

    db_path = os.path.join(args.storage_path, "task_scheduler.sqlite")
    faucet_seed, faucet_pub = read_solana_keypair_file(args.faucet_keypair)

    try:
        migrate(db_path, args.rpc_url, faucet_seed, faucet_pub, args.block_time_ms)
    except RpcError as e:
        log(f"Task migration failed: {e}")
        sys.exit(1)


if __name__ == "__main__":
    main()

Summary by CodeRabbit

• New Features
  • Task scheduling now runs via Hydra cranks (schedule, reschedule, cancel, and legacy migration).
  • Added an independently funded faucet account with optional startup delegation.
  • Added an account eviction instruction for ephemeral accounts.
  • Cranks are isolated per authority.
• Bug Fixes
  • Startup is more resilient: scheduler initialization/migration failures no longer block the validator.
  • Helps prevent draining validator funds by routing crank funding through the faucet.
• Configuration
  • Scheduler config/example now exposes only the faucet keypair (timing/retention controls removed).

[coderabbitai]

Note

Reviews paused

It looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the reviews.auto_review.auto_pause_after_reviewed_commits setting.

Use the following commands to manage reviews:

• @coderabbitai resume to resume automatic reviews.
• @coderabbitai review to trigger a single review.

Use the checkboxes below for quick actions:

• ▶️ Resume reviews
• 🔍 Trigger review

📝 Walkthrough

Walkthrough

The task scheduler is migrated from internal crank execution to Hydra-backed crank creation and cancellation. Scheduler configuration now contains a faucet keypair, persistence is limited to legacy-task migration, and validator startup delegates the faucet on-chain. Internal crank instruction and builtin paths are removed. Dependencies and integration tests are updated for Hydra crank lifecycle validation.

Assessment against linked issues:

Objective	Addressed	Explanation
Migrate cranks to an external Hydra service [#1375]	✅

Suggested reviewers: gabrielepicco, taco-paco, thlorenz

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch dode/crank-refactor

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[coderabbitai]

Actionable comments posted: 5

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)

magicblock-task-scheduler/src/db.rs (1)

117-143: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win

Don’t let one malformed row abort all migration.

A single bad instructions blob or authority string makes get_tasks() return Err, so valid tasks are not migrated and the DB is not drained. Parse rows independently and remove/quarantine only the malformed task id.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@magicblock-task-scheduler/src/db.rs` around lines 117 - 143, In the
task-loading path around the `stmt.query_map`/`rows.collect` logic in
`get_tasks`, a single malformed `instructions` blob or `authority` value
currently turns the whole read into `Err` and blocks draining valid tasks.
Change the row handling so each record is parsed independently, and on
deserialization/parse failure, skip or quarantine only that task’s `id` instead
of propagating the error. Keep the successful `DbTask` construction path
unchanged, but ensure bad rows are removed or marked separately so migration can
continue for the rest.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@magicblock-task-scheduler/src/db.rs`:
- Around line 19-28: Preserve the legacy scheduling timing when migrating tasks:
DbTask and get_tasks currently drop the previous last_execution_millis/timestamp
data, so recreated Hydra cranks can drift from their original cadence. Update
DbTask to retain enough legacy timing metadata and have get_tasks use it to
compute the correct Hydra start_slot, or otherwise make the migration explicitly
reset timing with documented behavior. Use the existing DbTask and get_tasks
flow as the place to carry this state through the migration path.

In `@magicblock-task-scheduler/src/hydra.rs`:
- Around line 67-94: The Hydra create serialization path in
CreateArgs::serialize and build_create_ix currently assumes scheduled is
non-empty and blindly casts meta/data lengths with as, which can generate
invalid wire data. Update the builder/serializer to return a Result and validate
the Hydra invariants up front: reject empty scheduled slices, and use checked
conversions for metas.len() and data.len() so oversized values fail
deterministically instead of truncating.

In `@magicblock-task-scheduler/src/service.rs`:
- Around line 143-158: Keep valid tasks in SQLite until Hydra creation succeeds;
the migration path in the task scheduler currently removes tasks even when
`wait_for_block_ready()` times out or `schedule_crank()` fails. Update the
migration logic around `schedule_crank`, `wait_for_block_ready`, and
`db.remove_task` so each task is only deleted after a successful Hydra creation,
and leave the SQLite row intact on any error so it can be retried later.
- Around line 201-213: Reject non-positive iteration counts before calling
schedule_crank so we never create a Hydra crank with remaining = 0; update the
task scheduling path in service.rs to validate task.iterations up front and
return early for zero or negative values, mirroring the migration behavior that
skips exhausted tasks. Apply the same guard in the other task creation/update
paths that feed schedule_crank, using the existing schedule_crank and
task.iterations symbols to locate the affected flows.

In `@programs/magicblock/src/schedule_task/mod.rs`:
- Around line 22-28: The signer rejection in the schedule_task logic is now
broader than the current log message implies, so update the stale error text in
the `if account.is_signer` branch to match the new rule. Keep the change
localized to the `ic_msg!` call in `schedule_task` so it clearly states that no
signer is allowed, rather than saying the crank signer PDA is permitted.

---

Outside diff comments:
In `@magicblock-task-scheduler/src/db.rs`:
- Around line 117-143: In the task-loading path around the
`stmt.query_map`/`rows.collect` logic in `get_tasks`, a single malformed
`instructions` blob or `authority` value currently turns the whole read into
`Err` and blocks draining valid tasks. Change the row handling so each record is
parsed independently, and on deserialization/parse failure, skip or quarantine
only that task’s `id` instead of propagating the error. Keep the successful
`DbTask` construction path unchanged, but ensure bad rows are removed or marked
separately so migration can continue for the rest.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: ASSERTIVE

Plan: Pro

Run ID: 70d07510-06dd-49d4-a882-8e1ad7b04ea5

📥 Commits

Reviewing files that changed from the base of the PR and between 5e8a24a and 5fe3626.

⛔ Files ignored due to path filters (3)

• Cargo.lock is excluded by !**/*.lock
• test-integration/Cargo.lock is excluded by !**/*.lock
• test-integration/programs/hydra/hydra.so is excluded by !**/*.so

📒 Files selected for processing (32)

• config.example.toml
• magicblock-api/src/magic_validator.rs
• magicblock-config/src/config/mod.rs
• magicblock-config/src/config/scheduler.rs
• magicblock-config/src/consts.rs
• magicblock-config/src/lib.rs
• magicblock-config/src/tests.rs
• magicblock-magic-program-api/src/instruction.rs
• magicblock-magic-program-api/src/pda.rs
• magicblock-processor/src/builtins.rs
• magicblock-task-scheduler/Cargo.toml
• magicblock-task-scheduler/src/db.rs
• magicblock-task-scheduler/src/hydra.rs
• magicblock-task-scheduler/src/lib.rs
• magicblock-task-scheduler/src/service.rs
• programs/magicblock/src/magicblock_processor.rs
• programs/magicblock/src/schedule_task/mod.rs
• programs/magicblock/src/schedule_task/process_execute_task.rs
• programs/magicblock/src/schedule_task/process_schedule_task.rs
• programs/magicblock/src/utils/instruction_utils.rs
• test-integration/configs/schedule-task.devnet.toml
• test-integration/test-ledger-restore/src/lib.rs
• test-integration/test-task-scheduler/src/lib.rs
• test-integration/test-task-scheduler/tests/test_cancel_ongoing_task.rs
• test-integration/test-task-scheduler/tests/test_independent_authority.rs
• test-integration/test-task-scheduler/tests/test_migration.rs
• test-integration/test-task-scheduler/tests/test_reschedule_task.rs
• test-integration/test-task-scheduler/tests/test_schedule_error.rs
• test-integration/test-task-scheduler/tests/test_schedule_magic_cpi_crank.rs
• test-integration/test-task-scheduler/tests/test_schedule_task.rs
• test-integration/test-task-scheduler/tests/test_unauthorized_reschedule.rs
• test-integration/test-task-scheduler/tests/test_use_crank_signer.rs

💤 Files with no reviewable changes (14)

• test-integration/test-task-scheduler/tests/test_use_crank_signer.rs
• test-integration/test-task-scheduler/tests/test_schedule_magic_cpi_crank.rs
• test-integration/test-task-scheduler/tests/test_schedule_error.rs
• magicblock-config/src/config/scheduler.rs
• magicblock-magic-program-api/src/pda.rs
• magicblock-processor/src/builtins.rs
• magicblock-config/src/consts.rs
• magicblock-magic-program-api/src/instruction.rs
• magicblock-api/src/magic_validator.rs
• test-integration/test-task-scheduler/tests/test_unauthorized_reschedule.rs
• config.example.toml
• magicblock-config/src/config/mod.rs
• programs/magicblock/src/schedule_task/process_execute_task.rs
• magicblock-config/src/tests.rs

[coderabbitai]

Actionable comments posted: 6

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (3)

magicblock-task-scheduler/src/service.rs (1)

110-125: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick win

Wait for a usable blockhash before serving runtime requests.

send_create() and send_cancel() pull self.block.load().blockhash directly, but the main loop only gates on faucet readiness. In magicblock-api/src/magic_validator.rs, the scheduler is spawned before init_slot_ticker(...), so the first schedule/cancel requests can still hit the default blockhash and get logged+dropped instead of retried. Gate the runtime loop on block readiness too, or retry requests until a real blockhash exists.

Also applies to: 217-230, 333-335, 391-399

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@magicblock-task-scheduler/src/service.rs` around lines 110 - 125, The
scheduler loop currently waits only for faucet readiness, but runtime requests
in send_create and send_cancel can still see the default blockhash and get
dropped. Update TaskSchedulerService::run (or the request processing path used
by process_request) to also wait until self.block.load().blockhash is usable
before receiving scheduled_tasks, or add retry/requeue handling when the
blockhash is not yet initialized. Make sure the same readiness guard covers the
initial spawn path from magic_validator so requests are not processed until both
faucet and block readiness are satisfied.

magicblock-api/src/magic_validator.rs (1)

439-442: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick win

Replace this expect with proper error propagation.

A missing tasks_service channel now panics during startup instead of returning a structured ApiError. Handle it like the replication-channel branch above and fail initialization cleanly.

Suggested change

         let task_scheduler = TaskSchedulerService::new(
             &task_scheduler_db_path,
             config.aperture.listen.http(),
             faucet_keypair.insecure_clone(),
             dispatch
                 .tasks_service
                 .take()
-                .expect("tasks_service should be initialized"),
+                .ok_or_else(|| {
+                    ApiError::FailedToSendModeSwitch(
+                        "tasks_service channel missing after init".to_owned(),
+                    )
+                })?,
             ledger.latest_block().clone(),
             Duration::from_millis(config.ledger.block_time_ms()),
             token.clone(),

As per path instructions, {magicblock-*,programs,storage-proto}/**: Treat any usage of .unwrap() or .expect() in production Rust code as a MAJOR issue. These should not be categorized as trivial or nit-level concerns. Request proper error handling or explicit justification with invariants.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@magicblock-api/src/magic_validator.rs` around lines 439 - 442, The
`tasks_service` initialization in `magic_validator.rs` still uses `expect`,
which can panic during startup instead of returning a structured `ApiError`.
Update the `tasks_service` branch in the same way as the replication-channel
handling above by checking the `Option`, converting the missing-channel case
into an error, and propagating it through the existing initialization flow. Use
the surrounding `dispatch.tasks_service.take()` logic and related startup error
path to keep failure handling consistent and non-panicking.

Source: Path instructions

test-integration/test-task-scheduler/tests/test_cancel_ongoing_task.rs (1)

72-81: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick win

Fetch a fresh blockhash before the cancel transaction.

wait_for_hydra_crank() can spend up to 10s here, but the cancel send below still reuses the blockhash fetched before scheduling. With the 50ms test block time, that can age out of the recent-blockhash window and make this test intermittently fail.

Proposed fix

     wait_for_hydra_crank(
         &ctx,
         &crank_pda,
         Duration::from_secs(10),
         &mut validator,
     );
 
+    let ephem_blockhash =
+        expect!(ctx.try_get_latest_blockhash_ephem(), validator);
+
     // Cancel the task while it is still ongoing.
     let sig = expect!(
         ctx.send_transaction_ephem_with_preflight(

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@test-integration/test-task-scheduler/tests/test_cancel_ongoing_task.rs`
around lines 72 - 81, The cancel flow in test_cancel_ongoing_task is reusing an
old blockhash after wait_for_hydra_crank() can block for several seconds, which
may let the transaction fall out of the recent-blockhash window. Before
constructing and sending the cancel transaction, fetch a fresh blockhash in the
same place you build the cancel request (near the cancel task step) so the
transaction uses up-to-date cluster state. Refer to the cancel path around
wait_for_hydra_crank(), the cancel send logic, and the task cancellation
transaction setup in test_cancel_ongoing_task.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@magicblock-api/Cargo.toml`:
- Line 56: Enable the missing bincode feature for the solana-system-interface
dependency in the Cargo.toml entry used by magicblock-api. The fund_account path
relies on solana_system_interface::instruction::assign, so update the existing
workspace dependency declaration to include the bincode feature and keep the
dependency location intact. Refer to the solana-system-interface dependency line
and the fund_account.rs usage when making the change.

In `@magicblock-api/src/errors.rs`:
- Around line 39-40: The `FailedToDelegateFaucet` error text in `errors.rs` is
misleading because `ensure_faucet_delegated_on_chain` only checks/delegates a
pre-funded faucet and never funds it. Update the `#[error(...)]` message for
`FailedToDelegateFaucet(Pubkey, String)` to describe the correct remediation:
the faucet must already be funded before delegation, while keeping the faucet
pubkey and underlying error details in the message.

In `@magicblock-api/src/fund_account.rs`:
- Around line 94-97: The early delegation check in fund_account should not rely
only on account.owner == dlp_api::id(), because it can miss cases where the
faucet is delegated to a different validator. Update the already_delegated logic
to verify the delegation matches the target validator binding used by the later
delegation setup in this function, or return an error when the existing
delegation points elsewhere. Apply the same validator-specific validation in the
related delegation path around the later delegation checks so startup does not
skip re-delegation on a mismatched faucet.

In `@magicblock-config/src/config/scheduler.rs`:
- Around line 20-27: The default implementation of TaskSchedulerConfig should
not fall back to the repository-known faucet private key. Update
TaskSchedulerConfig::default and any config-loading path used by
MagicValidator::try_from_config so task-scheduler.faucet-keypair must be
explicitly provided for real deployments, or is only populated by an explicit
dev/test-only mode. Replace consts::DEFAULT_TASK_SCHEDULER_FAUCET_KEYPAIR with a
non-reusable placeholder/example value and ensure
SerdeKeypair/Keypair::from_base58_string are not used to silently enable the
published secret by default.

In `@test-integration/test-task-scheduler/src/lib.rs`:
- Around line 71-89: The test harness is pre-delegating the faucet in a way that
hides startup delegation behavior. Update airdrop_faucet to only fund the faucet
and remove the delegate_account_to_validator call there, so setup_validator*
starts with an undelegated faucet and exercises
ensure_faucet_delegated_on_chain. If needed, add a separate startup-focused test
that explicitly begins with an undelegated faucet and verifies the delegation
path.

In `@test-integration/test-task-scheduler/tests/test_undrained_validator.rs`:
- Around line 33-35: The test is using a hard-coded default validator keypair
literal instead of the validator identity from the harness, which can make the
balance check target the wrong account. Update the setup in
test_undrained_validator to source the validator pubkey from setup_validator()
or reuse the shared consts::DEFAULT_VALIDATOR_KEYPAIR, and then use that value
wherever validator_identity is checked.

---

Outside diff comments:
In `@magicblock-api/src/magic_validator.rs`:
- Around line 439-442: The `tasks_service` initialization in
`magic_validator.rs` still uses `expect`, which can panic during startup instead
of returning a structured `ApiError`. Update the `tasks_service` branch in the
same way as the replication-channel handling above by checking the `Option`,
converting the missing-channel case into an error, and propagating it through
the existing initialization flow. Use the surrounding
`dispatch.tasks_service.take()` logic and related startup error path to keep
failure handling consistent and non-panicking.

In `@magicblock-task-scheduler/src/service.rs`:
- Around line 110-125: The scheduler loop currently waits only for faucet
readiness, but runtime requests in send_create and send_cancel can still see the
default blockhash and get dropped. Update TaskSchedulerService::run (or the
request processing path used by process_request) to also wait until
self.block.load().blockhash is usable before receiving scheduled_tasks, or add
retry/requeue handling when the blockhash is not yet initialized. Make sure the
same readiness guard covers the initial spawn path from magic_validator so
requests are not processed until both faucet and block readiness are satisfied.

In `@test-integration/test-task-scheduler/tests/test_cancel_ongoing_task.rs`:
- Around line 72-81: The cancel flow in test_cancel_ongoing_task is reusing an
old blockhash after wait_for_hydra_crank() can block for several seconds, which
may let the transaction fall out of the recent-blockhash window. Before
constructing and sending the cancel transaction, fetch a fresh blockhash in the
same place you build the cancel request (near the cancel task step) so the
transaction uses up-to-date cluster state. Refer to the cancel path around
wait_for_hydra_crank(), the cancel send logic, and the task cancellation
transaction setup in test_cancel_ongoing_task.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: ASSERTIVE

Plan: Pro

Run ID: c6831594-1813-492f-998c-c519f321926d

📥 Commits

Reviewing files that changed from the base of the PR and between 5fe3626 and 7850f16.

⛔ Files ignored due to path filters (3)

• Cargo.lock is excluded by !**/*.lock
• test-integration/Cargo.lock is excluded by !**/*.lock
• test-integration/programs/hydra/hydra.so is excluded by !**/*.so

📒 Files selected for processing (23)

• Cargo.toml
• config.example.toml
• magicblock-api/Cargo.toml
• magicblock-api/src/errors.rs
• magicblock-api/src/fund_account.rs
• magicblock-api/src/magic_validator.rs
• magicblock-config/src/config/scheduler.rs
• magicblock-config/src/consts.rs
• magicblock-config/src/lib.rs
• magicblock-config/src/types/crypto.rs
• magicblock-task-scheduler/Cargo.toml
• magicblock-task-scheduler/src/errors.rs
• magicblock-task-scheduler/src/lib.rs
• magicblock-task-scheduler/src/service.rs
• test-integration/Cargo.toml
• test-integration/test-task-scheduler/Cargo.toml
• test-integration/test-task-scheduler/src/lib.rs
• test-integration/test-task-scheduler/tests/test_cancel_ongoing_task.rs
• test-integration/test-task-scheduler/tests/test_independent_authority.rs
• test-integration/test-task-scheduler/tests/test_migration.rs
• test-integration/test-task-scheduler/tests/test_reschedule_task.rs
• test-integration/test-task-scheduler/tests/test_schedule_task.rs
• test-integration/test-task-scheduler/tests/test_undrained_validator.rs

💤 Files with no reviewable changes (1)

• magicblock-task-scheduler/src/lib.rs

[coderabbitai]

Actionable comments posted: 7

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)

magicblock-config/src/tests.rs (1)

581-590: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Add environment parsing coverage for the faucet keypair.

The replacement MBV_TASK_SCHEDULER__FAUCET_KEYPAIR contract is untested. Add a guard and assertions for explicit configuration and omission behavior.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@magicblock-config/src/tests.rs` around lines 581 - 590, Extend the
environment parsing tests around the existing EnvVarGuard setup to cover
MBV_TASK_SCHEDULER__FAUCET_KEYPAIR: add a guard with an explicit keypair value
and assert that configuration parses to that value, then verify the
omitted-variable case produces the expected absence/default behavior. Ensure the
guard is scoped so both explicit and omission assertions are isolated.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@magicblock-api/src/magic_validator.rs`:
- Around line 452-467: Update the validator initialization flow around
TaskSchedulerService::new to propagate initialization failures as an ApiError
instead of converting them with .ok(). Replace the changed
dispatch.tasks_service.expect(...) with explicit error propagation, preserving
the existing error context where appropriate, so start() never receives a
missing task scheduler and does not rely on an unconditional expect.
- Around line 971-989: Ensure scheduler startup cannot proceed until
ensure_faucet_delegated_on_chain completes successfully, rather than performing
the check only in the detached startup task. Add an explicit readiness barrier
or await this setup before the scheduler launch flow in start(), while
preserving the existing failure logging and process exit behavior.

In `@magicblock-task-scheduler/src/service.rs`:
- Around line 238-251: Update process_request and the schedule/cancel handling
so failed operations are not permanently dropped after a single Hydra/RPC error.
Persist pending operations or implement bounded retries that reconcile current
task state before retrying, especially for ambiguous create and cancel
transactions; preserve clear failure reporting after recovery attempts are
exhausted.
- Around line 134-137: Update the migration documentation to state that invalid
rows and successfully migrated rows are removed, while valid rows whose Hydra
crank creation fails remain in SQLite for a later startup. Apply this correction
in magicblock-task-scheduler/src/service.rs at lines 134-137 and align the
migration narrative in .agents/context/crates/magicblock-task-scheduler.md at
line 119; no code behavior changes are required.
- Around line 321-327: Update crank_exists to return TaskSchedulerResult<bool>,
preserving true only when the account owner is EPHEMERAL_PROGRAM_ID and
returning false only for a confirmed account-not-found response. Propagate all
other RPC failures instead of treating them as absent, and update callers to
handle the propagated result so reschedule cancellation and crank creation
remain correct.

In `@test-integration/test-task-scheduler/tests/test_reschedule_task.rs`:
- Around line 106-113: Strengthen the rescheduling assertion around the second
wait_for_hydra_crank call by capturing the crank account’s original lamports and
data before rescheduling, then polling until the account reflects the expected
larger funding or replacement state. Ensure the test fails when rescheduling is
ignored, rather than only confirming that the same crank PDA still exists.

In `@test-integration/test-task-scheduler/tests/test_schedule_task.rs`:
- Around line 100-106: Update wait_for_hydra_crank_closed and its callers so
RPC/transport failures propagate instead of being interpreted as closure;
explicitly treat account-not-found as closed. In
test-integration/test-task-scheduler/tests/test_schedule_task.rs:100-106, use
the corrected closure helper; in
test-integration/test-task-scheduler/tests/test_cancel_ongoing_task.rs:108-114,
verify account removal or zero lamports; in
test-integration/test-task-scheduler/tests/test_independent_authority.rs:117-122
and test-integration/test-task-scheduler/tests/test_reschedule_task.rs:137-142,
ensure transport errors cannot satisfy cancellation.

---

Outside diff comments:
In `@magicblock-config/src/tests.rs`:
- Around line 581-590: Extend the environment parsing tests around the existing
EnvVarGuard setup to cover MBV_TASK_SCHEDULER__FAUCET_KEYPAIR: add a guard with
an explicit keypair value and assert that configuration parses to that value,
then verify the omitted-variable case produces the expected absence/default
behavior. Ensure the guard is scoped so both explicit and omission assertions
are isolated.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: ASSERTIVE

Plan: Pro

Run ID: 042fcb77-e76b-41aa-a013-e6851291fb12

📥 Commits

Reviewing files that changed from the base of the PR and between 7850f16 and 9c1d10a.

⛔ Files ignored due to path filters (2)

• Cargo.lock is excluded by !**/*.lock
• test-integration/Cargo.lock is excluded by !**/*.lock

📒 Files selected for processing (29)

• .agents/context/crates/magicblock-task-scheduler.md
• Cargo.toml
• config.example.toml
• magicblock-api/Cargo.toml
• magicblock-api/src/crank_faucet.rs
• magicblock-api/src/errors.rs
• magicblock-api/src/lib.rs
• magicblock-api/src/magic_validator.rs
• magicblock-config/src/config/scheduler.rs
• magicblock-config/src/consts.rs
• magicblock-config/src/tests.rs
• magicblock-magic-program-api/src/instruction.rs
• magicblock-processor/src/builtins.rs
• magicblock-task-scheduler/src/db.rs
• magicblock-task-scheduler/src/service.rs
• programs/magicblock/src/magicblock_processor.rs
• programs/magicblock/src/schedule_task/mod.rs
• programs/magicblock/src/utils/instruction_utils.rs
• test-integration/Cargo.toml
• test-integration/configs/schedule-task.devnet.toml
• test-integration/test-task-scheduler/src/lib.rs
• test-integration/test-task-scheduler/tests/test_cancel_ongoing_task.rs
• test-integration/test-task-scheduler/tests/test_independent_authority.rs
• test-integration/test-task-scheduler/tests/test_migration.rs
• test-integration/test-task-scheduler/tests/test_reschedule_task.rs
• test-integration/test-task-scheduler/tests/test_schedule_task.rs
• test-integration/test-task-scheduler/tests/test_schedule_task_signed.rs
• test-integration/test-task-scheduler/tests/test_scheduled_commits.rs
• test-integration/test-task-scheduler/tests/test_undrained_faucet.rs

💤 Files with no reviewable changes (3)

• test-integration/test-task-scheduler/tests/test_scheduled_commits.rs
• magicblock-magic-program-api/src/instruction.rs
• magicblock-processor/src/builtins.rs

[coderabbitai]

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)

magicblock-task-scheduler/src/service.rs (1)

120-127: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value

Break the loop if the task channel is closed.

Currently, if the scheduled_tasks channel is closed (recv() returns None), the select! macro will disable the branch and block indefinitely on self.token.cancelled(). It is more idiomatic to exit the loop when the channel is closed to prevent the task from hanging unnecessarily.

♻️ Proposed refactor

             select! {
-                Some(request) = self.scheduled_tasks.recv() => {
-                    self.process_request(request).await;
-                }
+                request = self.scheduled_tasks.recv() => {
+                    match request {
+                        Some(req) => self.process_request(req).await,
+                        None => break,
+                    }
+                }
                 _ = self.token.cancelled() => {

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@magicblock-task-scheduler/src/service.rs` around lines 120 - 127, Update the
service loop’s select! branch receiving from self.scheduled_tasks to handle both
Some(request) and None, processing requests as before and breaking the loop when
the channel is closed; retain the existing self.token.cancelled() shutdown
behavior.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Outside diff comments:
In `@magicblock-task-scheduler/src/service.rs`:
- Around line 120-127: Update the service loop’s select! branch receiving from
self.scheduled_tasks to handle both Some(request) and None, processing requests
as before and breaking the loop when the channel is closed; retain the existing
self.token.cancelled() shutdown behavior.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: ASSERTIVE

Plan: Pro

Run ID: 93c27f68-337c-4267-8758-8f43d259952e

📥 Commits

Reviewing files that changed from the base of the PR and between 9c1d10a and 4a8d872.

📒 Files selected for processing (1)

• magicblock-task-scheduler/src/service.rs

[coderabbitai]

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)

magicblock-task-scheduler/src/service.rs (1)

120-128: 🩺 Stability & Availability | 🔴 Critical | ⚡ Quick win

Prevent panic on closed channel in tokio::select!.

If self.scheduled_tasks.recv() returns None (which happens if the channel is closed), the pattern Some(request) will fail to match. Because this tokio::select! macro lacks an else branch, a pattern mismatch will cause it to panic and crash the task scheduler.

Please handle the None case gracefully (e.g., by breaking out of the loop) to ensure stability.

💻 Proposed fix

         loop {
             select! {
-                Some(request) = self.scheduled_tasks.recv() => {
-                    self.process_request(request).await;
-                }
+                request_opt = self.scheduled_tasks.recv() => {
+                    match request_opt {
+                        Some(request) => self.process_request(request).await,
+                        None => {
+                            debug!("Scheduled tasks channel closed");
+                            break;
+                        }
+                    }
+                }
                 _ = self.token.cancelled() => {
                     break;
                 }

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@magicblock-task-scheduler/src/service.rs` around lines 120 - 128, Update the
tokio::select! loop around self.scheduled_tasks.recv() to handle a closed
channel when recv returns None, breaking out of the scheduler loop instead of
allowing the unmatched pattern to panic; preserve the existing request
processing and cancellation behavior.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Outside diff comments:
In `@magicblock-task-scheduler/src/service.rs`:
- Around line 120-128: Update the tokio::select! loop around
self.scheduled_tasks.recv() to handle a closed channel when recv returns None,
breaking out of the scheduler loop instead of allowing the unmatched pattern to
panic; preserve the existing request processing and cancellation behavior.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: ASSERTIVE

Plan: Pro

Run ID: 6ac21ada-ec5c-45ce-8591-652290664e70

📥 Commits

Reviewing files that changed from the base of the PR and between 4a8d872 and bccbdfa.

📒 Files selected for processing (5)

• .agents/context/crates/magicblock-task-scheduler.md
• magicblock-api/src/magic_validator.rs
• magicblock-task-scheduler/src/service.rs
• test-integration/test-task-scheduler/src/lib.rs
• test-integration/test-task-scheduler/tests/test_reschedule_task.rs

💤 Files with no reviewable changes (1)

• test-integration/test-task-scheduler/tests/test_reschedule_task.rs