refactor(consensus): Delete Engine Task Queue

[refcell]

Summary

Delete the remaining engine priority queue, task enum, trait dispatch, enqueue, drain, and delegated/finalize wrapper structs. The engine processor now publishes direct-operation state changes without a pre-drain step, and Engine is a non-generic state owner with generic methods for the client-specific calls. The legacy dev queue-length RPC now reports zero without carrying queue state through the engine actor.

[vercel]

The latest updates on your projects. Learn more about Vercel for GitHub.

1 Skipped Deployment

Project	Deployment	Actions	Updated (UTC)
base	Ignored	Preview	May 14, 2026 10:38pm

[github-actions]

The sender (_) from watch::channel(0) is immediately dropped. When the dev RPC subscriber calls wait_for on this receiver (in dev.rs:80), it will get Err(RecvError) immediately because the sender is gone, causing the subscription loop to exit and log a "Subscription to engine queue size has been closed" warning on every subscription attempt.

If this is intentional stub behavior for a deprecated API, consider either:

1. Documenting why the sender is intentionally dropped (e.g., a comment explaining the legacy stub), or
2. Keeping the sender alive so the subscription hangs (returning 0 forever) rather than erroring out — which might be friendlier to existing callers expecting a live subscription.

[github-actions]

engine_task_count is incremented unconditionally here (on both success and failure), but the metric is described as "Engine operations successfully executed" and retry_with_severity only increments it on success (line 294). This will inflate the success counter for failed local inserts.

Suggested change

	Metrics::engine_task_count(Metrics::INSERT_TASK_LABEL).increment(1);
	if result.is_ok() {
	Metrics::engine_task_count(Metrics::INSERT_TASK_LABEL).increment(1);
	}

[github-actions]

Review Summary

Clean refactor that removes the engine task queue abstraction in favor of direct operations on Engine. The Engine struct is now non-generic (no PhantomData<EngineClient_>), with generics pushed to individual methods — a nice simplification. The EngineTaskExt trait, SealTask, BuildAndSealError, EngineTaskErrors, and build_and_seal utility are removed; their logic is inlined as Engine methods (seal_started_payload_with_state, insert_sealed_payload_with_state, rebuild_safe_payload). The EngineOperationError enum is correctly moved to the service crate where it belongs (the engine crate no longer needs to know about the aggregate error type).

Issues (2 findings, both covered by existing inline comments)

1. Metrics inflation in insert_local_unsafe_payload (operations/core.rs:218): engine_task_count is incremented unconditionally (success and failure), unlike retry_with_severity which only increments on success. This inflates the "operations successfully executed" counter on failed local inserts.

2. Immediately-dropped watch sender in QueueLengthReceiver shim (query.rs:139): let (_, queue_length_recv) = watch::channel(0) drops the sender, causing any wait_for caller to receive Err(RecvError) immediately. The TaskQueueLength shim (returns 0) is fine, but the subscription variant will error out on each subscription attempt rather than hanging with the shim value.

No new findings beyond the existing inline comments.