hive-c0re: serialise the matrix and knowledge sweeps through the job queue

The matrix sweep and the /knowledge pull each had concurrent callers
(#4723 item 4). Two overlapping knowledge pulls fail on .git/index.lock
and the remote-tracking ref lock: 30 of 30 concurrent replays of the
reset/clean/pull sequence in a scratch repo errored, 0 of 10 sequential
ones did. Two overlapping matrix sweeps on a hive with no persisted
Space / chat-room id both miss the by-name lookup and both createRoom
(from reading the code, not reproduced against a homeserver). On every
boot the MatrixSweep DAG node and the main.rs loop's immediate first
call ran at once.

Every sweep now runs as a job node, and each sweep's node holds its own
capacity-1 queue resource (Resource::MatrixSweep,
Resource::KnowledgeTree), the MetaWindow pattern: the scheduler never
starts a second pass of one sweep while the first holds the resource,
and different sweeps still run side by side.

- templates::matrix_sweep / templates::knowledge_pull build the node
  with its resource; boot, the periodic loops and the swarm event all
  use them.
- JobQueue::insert_unless_live folds a submission into a live node of
  the same kind instead of queueing another. Periodic ticks fold into a
  queued or running pass. The swarm knowledge event folds into a queued
  pull only, and queues one behind a running pull, which may have
  fetched before the push.
- The main.rs matrix loop no longer sweeps immediately at startup; the
  boot MatrixSweep node is the startup pass, as KnowledgePull already
  was for knowledge.
- The executors bound each pass (10 min matrix, 5 min knowledge), since
  a hung pass would otherwise hold its resource against every later one,
  and own the sweep-health banners, so every pass reports to them.

Replaces the SweepLock version of this branch, per review.

Refs #4723

hive-c0re/src/job_queue/scheduler.rs

@@ -205,3 +205,104 @@ fn reconcile_transients(coord: &Arc<Coordinator>, prev: &mut TransientSeen) {
         prev.insert(key, t.takes_container_down);
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use hive_jobq::scheduler::{Outcome, Scheduler};
+
+    use super::super::{JobQueue, NodeKind, State, templates};
+
+    /// Claim one runnable node through the same seam [`super::run_worker`]
+    /// uses. The returned future completes the node when awaited. `None` when
+    /// nothing is runnable.
+    fn claim(q: &JobQueue) -> Option<impl Future<Output = ()>> {
+        Scheduler::claim_next(q.sched(), |_, _, builder| async move {
+            (builder, Outcome::Done)
+        })
+        .map(|run| async move {
+            run.await.1.expect("a sweep grows nothing");
+        })
+    }
+
+    /// Kinds of the nodes currently `Running`, sorted.
+    fn running(q: &JobQueue) -> Vec<&'static str> {
+        let sched = q.sched().lock().expect("job_queue mutex poisoned");
+        let mut kinds: Vec<&'static str> = sched
+            .graph()
+            .nodes()
+            .filter(|n| n.state == State::Running)
+            .map(|n| <&'static str>::from(&n.payload))
+            .collect();
+        kinds.sort_unstable();
+        kinds
+    }
+
+    fn count(q: &JobQueue, kind: &str) -> usize {
+        let sched = q.sched().lock().expect("job_queue mutex poisoned");
+        sched
+            .graph()
+            .nodes()
+            .filter(|n| <&str>::from(&n.payload) == kind)
+            .count()
+    }
+
+    #[tokio::test]
+    async fn two_passes_of_one_sweep_never_run_together() {
+        let q = JobQueue::new(1);
+        for _ in 0..2 {
+            q.insert_job(|b| vec![templates::matrix_sweep(b).guid()])
+                .expect("insert");
+        }
+        q.insert_job(|b| vec![templates::knowledge_pull(b).guid()])
+            .expect("insert");
+
+        let first = claim(&q).expect("a matrix pass is runnable");
+        let other = claim(&q).expect("a different sweep runs alongside it");
+        assert!(
+            claim(&q).is_none(),
+            "the second matrix pass waits for the first"
+        );
+        assert_eq!(running(&q), ["knowledge_pull", "matrix_sweep"]);
+
+        first.await;
+        other.await;
+        let second = claim(&q).expect("the second matrix pass runs once the first is done");
+        assert_eq!(running(&q), ["matrix_sweep"]);
+        second.await;
+    }
+
+    #[tokio::test]
+    async fn a_tick_folds_into_a_live_pull_and_an_event_queues_one_behind_it() {
+        let q = JobQueue::new(1);
+        let kind = NodeKind::KnowledgePull;
+        let queued = [State::Pending];
+        let live = [State::Pending, State::Running];
+        let submit = |fold_into: &[State]| {
+            q.insert_unless_live(&kind, fold_into, templates::knowledge_pull)
+                .expect("insert")
+                .is_some()
+        };
+
+        assert!(submit(&live), "nothing live: a tick inserts");
+        let pull = claim(&q).expect("the pull is runnable");
+        assert!(!submit(&live), "a tick folds into the running pull");
+        assert!(
+            submit(&queued),
+            "an event queues a pull behind the running one"
+        );
+        assert!(
+            !submit(&queued),
+            "a second event folds into the queued pull"
+        );
+        assert!(!submit(&live), "a tick folds into the queued pull");
+        assert_eq!(count(&q, "knowledge_pull"), 2);
+
+        assert!(
+            q.insert_unless_live(&NodeKind::MatrixSweep, &live, templates::matrix_sweep)
+                .expect("insert")
+                .is_some(),
+            "a live pull does not fold a different sweep"
+        );
+        pull.await;
+    }
+}