refactor(spider-scheduler): Move dispatch queue ownership into the scheduler core.

[LinZhihao-723]

Description

The scheduler runtime currently creates the dispatch queue, splits it into a (writer, reader) pair, hands the writer to the core as a parameter of SchedulerCore::run, and keeps the reader for the execution-manager-facing service. That arrangement assumes the dispatch queue is one channel, and two consequences follow from the assumption:

• The runtime must know the sink's concrete type at compile time. create_runtime calls make_core::<SchedulerStorageClientType, DispatchQueueWriter>(), so every core implementation is forced to share one dispatch queue type.
• A core cannot own a dispatch structure that is not a single channel. A resource-group-aware core's dispatch state is a set of per-group queues created on demand by either side, a hint channel, and a table co-owned with the service. There is no way to build that outside the core and pass it in as a sink.

This PR inverts the ownership: the core constructs and owns whatever dispatch structure its algorithm needs, and exposes only a read handle to the service. The runtime builds the core first, takes the handle from it, and then builds the service.

let core = scheduler_config.make_core::<SchedulerStorageClientType>();
let dispatch_queue_handle = core.get_dispatch_queue_handle();
let service = SchedulerServiceState::new(dispatch_queue_handle, registry, scheduler_id);

API changes

Before	After
SchedulerCore::Sink (associated type)	removed
SchedulerCore::run(self, storage_client, sink, ..)	SchedulerCore::run(self, storage_client, ..)
—	SchedulerCore::get_dispatch_queue_handle(&self) -> SharedDispatchQueueHandle
trait DispatchQueueSource: Send + Sync + Clone	trait DispatchQueueHandle: Send + Sync
—	pub type SharedDispatchQueueHandle = Arc<dyn DispatchQueueHandle>
trait DispatchQueueSink	removed; enqueue, bump_session_id and size become inherent methods on DispatchQueueWriter
SchedulerServiceState<DispatchQueueSourceType>	SchedulerServiceState
GrpcSchedulerService<DispatchQueueSourceType>	GrpcSchedulerService
SchedulerConfig::make_core<SchedulerStorageClientType, DispatchQueueSinkType>()	SchedulerConfig::make_core<SchedulerStorageClientType>()
SchedulerConfig::dispatch_queue_capacity()	removed
create_runtime(..) -> (Runtime, SchedulerServiceState<DispatchQueueReader>, CancellationToken)	create_runtime(..) -> (Runtime, SchedulerServiceState, CancellationToken)

Behaviour is unchanged. The queue is still created at SessionId::default() with the capacity from the core's own config, which is exactly what the runtime did before.

Why Arc and not Box

Clone implies Sized, so the trait could not be made into a trait object while Clone was a supertrait. Dropping it is what makes dyn DispatchQueueHandle possible; #[async_trait] boxes the returned future, so dequeue remains callable through the trait object.

The handle is an Arc with exactly one level of indirection. DispatchQueueReader previously held an Arc<DispatchQueueReaderInner>, so returning Arc<dyn DispatchQueueHandle> unchanged would have wrapped an Arc in an Arc — three pointer hops from SchedulerServiceState to the channel where there had been two. DispatchQueueReaderInner is therefore flattened into DispatchQueueReader, whose two remaining fields (session_id and assignment_receiver) are already cheap-clone handles.

Box was considered and rejected: it removes the double indirection, but hands the service a source with no guarantee of cheap cloning. Nothing clones it today, but that is a property of the current service rather than of the contract, and the contract is what a second core inherits.

Why DispatchQueueSink is deleted rather than kept

Before this change the trait was load-bearing: RoundRobinCore was generic over DispatchQueueSinkType: DispatchQueueSink, so the bound did real work. Once the core owns a concrete DispatchQueueWriter, the only holder is that concrete type, every call resolves statically, and the trait's sole remaining effect is forcing an import to bring the methods into scope.

The surviving trait stays a trait, and the asymmetry is the point: it is the seam the service depends on, and a second core will implement it differently. DispatchQueueSink had no second implementor coming.

With DispatchQueueSink gone, DispatchQueueSource named one half of a pair that no longer existed, so it is renamed to DispatchQueueHandle — the sole external access point to a core's dispatch queue, which is expected to grow further caller-to-core methods beyond dequeue. Its Arc alias follows the Shared<Thing> convention already used by spider-storage's SharedRw and SharedJobControlBlock.

One consequence worth flagging for review: size gained #[must_use]. As a trait method it was exempt from clippy::must_use_candidate; as a public inherent method it is not, and the pedantic lint group is deny-level in this workspace.

What deliberately does not change

These were considered and left alone, so please read them as decisions rather than as misses:

• DispatchQueueHandle::dequeue keeps its signature. A resource-group-aware core needs to distinguish a pinned execution manager from a general one, which dequeue(wait_time) cannot express. The minimal shape is dequeue(&self, resource_group_id: Option<ResourceGroupId>, wait_time: Duration), with the round-robin implementation ignoring the argument. Deferred to a follow-up PR to keep this diff reviewable.
• SchedulerCore: Send is retained, and create_runtime still spawns core.run(..) with tokio::spawn. A core whose scheduling state is not Send cannot be spawned that way; addressing it is a separate change and does not affect anything here.
• The round-robin scheduling policy. No admission, ordering or retirement behaviour is touched. The only change to RoundRobin is that it holds a concrete DispatchQueueWriter field instead of a generic sink, renamed to dispatch_queue_writer to match.
• create_dispatch_queue remains public. The round-robin core calls it from make_core, and the white-box tests call it directly.

Notes for reviewers

The round-robin tests already passed the real DispatchQueueWriter rather than a mock, so the test changes are mechanical: the helpers stop threading a sink parameter and take the core's reader for assertions instead.

RoundRobin::run destructures *self with .., which drops the core's dispatch queue reader. That is safe because DispatchQueueWriterInner holds its own clone of the async_channel::Receiver, so the channel does not close even if get_dispatch_queue_handle is never called.

Checklist

• The PR satisfies the contribution guidelines.
• This is a breaking change and that has been indicated in the PR title, OR this isn't a
  breaking change.
• Necessary docs have been updated, OR no docs need to be updated.

Validation performed

• Ensure all workflows pass.

Summary by CodeRabbit

• Refactor
  • Consolidated dispatch-queue ownership within the scheduler for simpler integration.
  • Introduced a single dispatch-queue handle for monitoring and retrieving scheduled work.
  • Simplified scheduler, runtime, and service interfaces.
  • Preserved queue behaviour, including back-pressure, session validation, task assignment, and rescheduling.
• Tests
  • Updated coverage for queue draining, rescheduling, deduplication, session updates, and job finalization.

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Repository UI

Review profile: CHILL

Plan: Pro Plus

Run ID: 809ad511-403f-46e5-8398-201c2f6647b8

📥 Commits

Reviewing files that changed from the base of the PR and between 61b17e5 and 5d55d6e.

📒 Files selected for processing (1)

• components/spider-scheduler/src/dispatch_queue.rs

💤 Files with no reviewable changes (1)

• components/spider-scheduler/src/dispatch_queue.rs

Included review availability: Your plan includes up to 1 review per rolling hour; 0 remain after this review.

---

Walkthrough

The scheduler now owns its dispatch queue. The API uses DispatchQueueHandle and SharedDispatchQueueHandle instead of separate sink and source traits. Runtime services consume the core-owned queue handle.

Changes

Dispatch queue ownership refactor

Layer / File(s)	Summary
Unified queue handle API components/spider-scheduler/src/dispatch_queue.rs	The separate sink and source traits were replaced with DispatchQueueHandle. Writer operations are now inherent methods.
Core-owned queue construction components/spider-scheduler/src/config.rs, components/spider-scheduler/src/core.rs, components/spider-scheduler/src/core_impl/round_robin/implementation.rs	SchedulerCore creates and owns the dispatch queue. RoundRobinCore exposes a shared handle and enqueues assignments through its writer.
Runtime and service integration components/spider-scheduler/src/runtime.rs, components/spider-scheduler/src/service.rs, components/spider-scheduler/src/grpc.rs, components/spider-scheduler/src/lib.rs	Runtime setup obtains the core-owned handle. Service and gRPC state no longer use generic dispatch-source types. Public exports and architecture documentation were updated.
Scheduler test updates components/spider-scheduler/src/core_impl/round_robin/tests.rs, components/spider-scheduler/src/service.rs	Tests use scheduler-owned handles for assignment draining and updated queue-handle test doubles.

Estimated code review effort: 4 (Complex) | ~45 minutes

Merge Risk: ⚪ Minimal · up to 5d55d

This change moves dispatch-queue ownership into the scheduler core while preserving scheduling behavior and service access through a read handle; no actionable merge-blocking risk remains beyond normal checks and review.

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title clearly and concisely describes the primary change: moving dispatch queue ownership into the scheduler core.
Docstring Coverage	✅ Passed	Docstring coverage is 100.00% which is sufficient. The required threshold is 80.00%.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests

---

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.}

[sitaowang1998]

LGTM.

[sitaowang1998]

Do we really need this docstring?

[LinZhihao-723]

Removed. I thought the docstring is necessary since we now make the API public. But looks like clippy doesn't complain if I remove it.

[sitaowang1998]

Verified that e2e tests from #385 pass after merge.