fix: gate outbox drain on holdership rather than local topic presence

Author: St4NNi

operations/src/metadata/forward.rs

@@ -423,10 +423,15 @@ pub(crate) async fn forward_outbox_record(
 ///
 /// The peer is a configured node of this realm and the record's bucket is one
 /// this node holds, so the record is queued into the local outbox verbatim and
-/// drains onto the bucket's topic like any locally-emitted record. Verbatim
-/// matters for admin operations: they are ordered by `(origin_node_id,
-/// origin_seq)` on every receiver, so re-stamping the relay as the origin would
-/// break that ordering and drop the emitter's earlier operations as stale.
+/// drains onto the bucket's topic like any locally-emitted record.
+///
+/// Document upserts and deletes only: an admin operation cannot be relayed at
+/// all. Receivers authenticate an admin event by requiring its signed publisher
+/// on the topic to *be* its `origin_node_id` (`validate_replicated_admin_event`),
+/// and a relay is by construction a different node signing for another origin, so
+/// every receiver would reject it. Accepting one here would report success while
+/// the operation was silently dropped realm-wide — worse than refusing it, which
+/// leaves the record in the emitter's outbox, retried and loud.
 async fn apply_forwarded_outbox_record(
     context: &Arc<DriverContext>,
     peer: NodeId,
@@ -446,6 +451,14 @@ async fn apply_forwarded_outbox_record(
         Ok(_) => {}
         Err(error) => return reject(error.to_string()),
     }
+    if matches!(
+        record.event,
+        aruna_core::document::DocumentSyncOutboxEvent::AdminOperation { .. }
+    ) {
+        return reject(
+            "an admin operation cannot be relayed: receivers require its signed publisher to be its origin node",
+        );
+    }
     if !holds_placement(&config, &record.placement, net_handle.node_id()) {
         return reject(format!(
             "node holds none of bucket {}/{} of the relayed outbox record",

operations/src/task_incoming.rs

@@ -214,19 +214,20 @@ impl DrainSyncOutcome {
 
 /// Per-run defer state, carried across the drain's pages so a topic deferred on
 /// one page keeps deferring on later pages (preserving FIFO within a topic) and
-/// each topic's genesis presence is probed at most once per run.
+/// each topic's holdership and genesis presence are decided at most once per run.
 #[derive(Default)]
 struct DrainDeferState {
     topic_exists: HashMap<irokle::TopicId, bool>,
+    topic_held: HashMap<irokle::TopicId, bool>,
     deferred_topics: HashSet<irokle::TopicId>,
     undeliverable_topics: HashSet<irokle::TopicId>,
 }
 
-/// What a record whose shard topic has no local genesis is waiting for.
+/// Whether a shard-classed record can ever publish from this node.
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 enum DeferOutcome {
-    /// The local node holds the bucket, so the genesis is on its way (rank-0
-    /// creates it, every other holder pulls it): keep the record and retry.
+    /// The local node holds the bucket, so it may publish onto the bucket's topic
+    /// once the genesis is there (rank-0 creates it, every other holder pulls it).
     Retry,
     /// The local node holds none of the record's bucket. Topic membership is the
     /// holder set, so it may neither mint the genesis nor join the topic: this
@@ -236,14 +237,22 @@ enum DeferOutcome {
 
 /// Splits FIFO-ordered drain records into those to publish now, those deferred
 /// because their shard topic has no local genesis yet, and those that can never
-/// publish from this node at all. Each topic's availability is evaluated once
-/// per run (state persists in `defer`); once a topic defers, every later record
-/// of that topic defers too. Splitting FIFO-adjacent records of one topic across
-/// a defer/publish boundary would let the newer op publish first, invert its
-/// origin sequence on receivers, and drop the older op forever as
-/// StaleOriginSequence — so a topic never straddles that boundary within a run,
-/// across pages included. A shard topic's holder set is a function of its
-/// placement, so undeliverability is likewise decided once per topic.
+/// publish from this node at all.
+///
+/// Holdership is checked *before* topic availability, and it is the only thing
+/// that decides publishability. Joining a shard topic is not holder-gated — a
+/// non-holder can adopt a co-holder's genesis, and the drain's own bootstrap pass
+/// will happily pull one — but its publishes onto that topic are not accepted. So
+/// "the topic exists locally" is not evidence this node may publish onto it, and
+/// classifying on that first would route a non-holder's records into a publish
+/// that silently goes nowhere instead of into the forwarding path.
+///
+/// Each topic's holdership and genesis presence are decided once per run (state
+/// persists in `defer`); once a topic defers, every later record of that topic
+/// defers too. Splitting FIFO-adjacent records of one topic across a
+/// defer/publish boundary would let the newer op publish first, invert its origin
+/// sequence on receivers, and drop the older op forever as StaleOriginSequence —
+/// so a topic never straddles that boundary within a run, across pages included.
 fn partition_drain_records(
     records: Vec<DrainRecord>,
     defer: &mut DrainDeferState,
@@ -262,6 +271,15 @@ fn partition_drain_records(
             undeliverable.push((record_key, record, topic));
             continue;
         }
+        let held = *defer
+            .topic_held
+            .entry(topic)
+            .or_insert_with(|| classify_defer(&record) == DeferOutcome::Retry);
+        if !held {
+            defer.undeliverable_topics.insert(topic);
+            undeliverable.push((record_key, record, topic));
+            continue;
+        }
         let already_deferred = defer.deferred_topics.contains(&topic);
         let available = !already_deferred
             && *defer
@@ -272,11 +290,6 @@ fn partition_drain_records(
             to_publish.push((record_key, record, topic));
             continue;
         }
-        if !already_deferred && classify_defer(&record) == DeferOutcome::Undeliverable {
-            defer.undeliverable_topics.insert(topic);
-            undeliverable.push((record_key, record, topic));
-            continue;
-        }
         defer.deferred_topics.insert(topic);
         debug!(
             event = "pipeline.drain.deferred",
@@ -557,11 +570,23 @@ impl OperationsTaskHandler {
             // short retry — the genesis arrives via gossip or the next pass. A
             // topic already deferred earlier this run is skipped: its records
             // stay deferred regardless, and re-probing would only waste RPCs.
+            //
+            // Only buckets this node holds are pulled. Joining is not holder-gated,
+            // so a non-holder could adopt the genesis here — and would then look
+            // publishable while its publishes went nowhere. Its records belong in
+            // the forwarding path instead.
             let mut missing_topics: BTreeMap<Vec<aruna_core::NodeId>, BTreeSet<irokle::TopicId>> =
                 BTreeMap::new();
             for (_, record, topic) in &records {
                 if !record.target.uses_shard_topic()
                     || defer_state.deferred_topics.contains(topic)
+                    || !realm_config.as_ref().is_none_or(|config| {
+                        crate::placement::holds_placement(
+                            config,
+                            &record.placement,
+                            net_handle.node_id(),
+                        )
+                    })
                     || net_handle
                         .document_sync_topic_exists(*topic)
                         .unwrap_or(false)

operations/tests/metadata_forwarding.rs

@@ -97,6 +97,67 @@ async fn user_node_forwards_create() -> Result<(), Box<dyn std::error::Error>> {
     Ok(())
 }
 
+/// A group created on a User-kind node is never silently lost.
+///
+/// A User node holds no bucket of any group, so its admin-operation outbox
+/// records can never publish from here. They also cannot be relayed: receivers
+/// authenticate an admin event by requiring its signed publisher to be its
+/// `origin_node_id`, so a holder publishing on the emitter's behalf is rejected
+/// realm-wide (`validate_replicated_admin_event`). The record therefore stays in
+/// the outbox, retried and loud, instead of being deleted after the caller was
+/// told the group exists. Creating groups from a User node needs a signed-origin
+/// admin relay in the net layer; until then this test pins the invariant that
+/// matters — the write is never thrown away.
+#[tokio::test]
+async fn user_node_group_survives() -> Result<(), Box<dyn std::error::Error>> {
+    let realm = Realm::new();
+    let (nodes, _config) = build_realm(&realm, 3, 1).await?;
+    let user_node = nodes.last().expect("user node");
+
+    let (group, _auth) = drive(
+        CreateGroupOperation::new(CreateGroupConfig {
+            actor: Actor {
+                node_id: user_node.net.node_id(),
+                user_id: realm.user_id,
+                realm_id: realm.realm_id,
+            },
+            display_name: "group from a user node".to_string(),
+            owner_cap: None,
+        }),
+        user_node.context.as_ref(),
+    )
+    .await?;
+
+    // Give the drain time to run: it must not have deleted the records.
+    sleep(Duration::from_secs(2)).await;
+    assert!(read_group_auth(user_node, group.group_id).await?.is_some());
+    assert!(
+        outbox_len(user_node).await? > 0,
+        "the user node's unpublishable admin records must be retained, not dropped"
+    );
+
+    shutdown(nodes).await;
+    Ok(())
+}
+
+async fn outbox_len(node: &TestNode) -> Result<usize, Box<dyn std::error::Error>> {
+    match node
+        .context
+        .storage_handle
+        .send_storage_effect(aruna_core::effects::StorageEffect::Iter {
+            key_space: aruna_core::keyspaces::DOCUMENT_SYNC_OUTBOX_KEYSPACE.to_string(),
+            prefix: None,
+            start: None,
+            limit: 64,
+            txn_id: None,
+        })
+        .await
+    {
+        Event::Storage(StorageEvent::IterResult { values, .. }) => Ok(values.len()),
+        other => Err(format!("unexpected outbox iter event: {other:?}").into()),
+    }
+}
+
 /// Re-forwarding a create that already applied returns the existing record.
 ///
 /// A forward whose response is lost is retried, possibly against a different
@@ -253,20 +314,27 @@ async fn seed_group(realm: &Realm, nodes: &[TestNode]) -> Result<Ulid, Box<dyn s
     )
     .await?;
 
+    wait_for_group(nodes, group.group_id).await?;
+    Ok(group.group_id)
+}
+
+/// Waits for the group's authorization document — what the permission check
+/// reads — on every sync-eligible node. A User node is never a sync target, so it
+/// never receives the group it may itself have created.
+async fn wait_for_group(
+    nodes: &[TestNode],
+    group_id: Ulid,
+) -> Result<(), Box<dyn std::error::Error>> {
     let deadline = Instant::now() + CONVERGENCE_TIMEOUT;
     loop {
         let mut pending = false;
-        for node in nodes {
-            // A user node is not a sync target, so it never receives the group.
-            if !node.sync_eligible {
-                continue;
-            }
-            if read_group_auth(node, group.group_id).await?.is_none() {
+        for node in nodes.iter().filter(|node| node.sync_eligible) {
+            if read_group_auth(node, group_id).await?.is_none() {
                 pending = true;
             }
         }
         if !pending {
-            return Ok(group.group_id);
+            return Ok(());
         }
         if Instant::now() >= deadline {
             return Err("the group's authorization document never reached every holder".into());