event_graph: Hide unchecked insertion paths

bin/darkirc/src/irc/client.rs

@@ -18,7 +18,6 @@
 
 use std::{
     collections::{HashMap, HashSet, VecDeque},
-    slice,
     sync::{
         atomic::{AtomicBool, AtomicUsize, Ordering::SeqCst},
         Arc,
@@ -255,47 +254,28 @@ impl Client {
                                     }
                                 };
 
-                                // Only now do we insert the header
-                                // and event into our local DAG. If
-                                // either insert fails we still log
-                                // and pass - the broadcast is a
-                                // best-effort fan-out anyway, and
-                                // the receiving peers' own validation
-                                // doesn't depend on our successful
-                                // local insert.
-                                if let Err(e) = self.server.darkirc.event_graph.header_dag_insert(vec![event.header.clone()], &dag_name).await {
-                                    error!("[IRC CLIENT] Failed inserting new header to Header DAG: {}", e);
-                                }
-                                if let Err(e) = self.server.darkirc.event_graph.dag_insert(slice::from_ref(&event), &dag_name).await {
-                                    error!("[IRC CLIENT] Failed inserting new event to DAG: {e}");
-                                    continue
-                                }
-
-                                // We sent this, so it should be considered seen.
-                                if let Err(e) = self.mark_seen(&event_id).await {
-                                    error!("[IRC CLIENT] (multiplex_connection) self.mark_seen({event_id}) failed: {e}");
-                                    return Err(e)
-                                }
-
-                                // Persist the blob locally so that a
-                                // late-joining peer EventReq-ing us
-                                // for this event gets a real blob in
-                                // the EventRep, not an empty one.
-                                // The receive-side path
-                                // (proto::handle_event_put) does the
-                                // analogous store after RLN
-                                // verification; this is the matching
-                                // step on the originator side.
+                                // Commit our outbound signal through
+                                // the safe public API. It inserts the
+                                // header, verifies and stores the RLN
+                                // blob, then commits the event body.
                                 if let Err(e) = self
                                     .server
                                     .darkirc
                                     .event_graph
-                                    .dag_blob_store(&event_id, &blob)
+                                    .insert_signal_with_blob(&event, &blob, &dag_name)
+                                    .await
                                 {
                                     error!(
-                                        "[IRC CLIENT] Failed persisting outbound \
-                                         RLN blob for {event_id}: {e}"
+                                        "[IRC CLIENT] Failed inserting verified \
+                                         signal event: {e}"
                                     );
+                                    continue
+                                }
+
+                                // We sent this, so it should be considered seen.
+                                if let Err(e) = self.mark_seen(&event_id).await {
+                                    error!("[IRC CLIENT] (multiplex_connection) self.mark_seen({event_id}) failed: {e}");
+                                    return Err(e)
                                 }
 
                                 self.server.darkirc.p2p.broadcast(&EventPut(event, blob)).await;

src/event_graph/mod.rs

@@ -1519,25 +1519,89 @@ impl EventGraph {
         }
     }
 
-    /// Insert events into a rotating DAG **without RLN verification**.
+    /// Public insertion path for a rotating-DAG signal event.
     ///
-    /// This is the post-verification entry point for callers that
-    /// have already verified the proof separately. Two legitimate
-    /// callers in production:
+    /// Non-genesis rotating events must carry an RLN signal blob. This method
+    /// inserts the header, re-verifies the blob, records RLN metadata, stores
+    /// the blob for future sync, and only then commits the event body. External
+    /// applications should use this instead of the unchecked post-verification
+    /// insertion path.
+    pub async fn insert_signal_with_blob(
+        &self,
+        event: &Event,
+        blob: &[u8],
+        dag_name: &str,
+    ) -> Result<Vec<blake3::Hash>> {
+        if event.header.parents != NULL_PARENTS && blob.is_empty() {
+            return Err(Error::Custom("rotating-DAG signal event blob must not be empty".into()))
+        }
+
+        let dag_ts = u64::from_str(dag_name)?;
+        let already_known = if event.header.parents == NULL_PARENTS {
+            false
+        } else {
+            let store = self.dag_store.read().await;
+            match store.get_slot(&dag_ts) {
+                Some(slot) => slot.main_tree.contains_key(event.id().as_bytes())?,
+                None => false,
+            }
+        };
+
+        self.header_dag_insert(vec![event.header.clone()], dag_name).await?;
+        let blobs = if blob.is_empty() { vec![] } else { vec![blob.to_vec()] };
+        let ids = self.dag_insert_with_blobs(std::slice::from_ref(event), &blobs, dag_name).await?;
+        let accepted = ids.contains(&event.id()) || already_known || {
+            let store = self.dag_store.read().await;
+            match store.get_slot(&dag_ts) {
+                Some(slot) => slot.main_tree.contains_key(event.id().as_bytes())?,
+                None => false,
+            }
+        };
+        if event.header.parents != NULL_PARENTS && !accepted {
+            return Err(Error::Custom("rotating-DAG signal event was not accepted".into()))
+        }
+
+        Ok(ids)
+    }
+
+    /// Insert events into a rotating DAG **without RLN verification**.
     ///
-    /// * `handle_event_put` - already ran `rln_verify_signal` and
-    ///   recorded the share. Calling `dag_insert_with_blobs` would
-    ///   trigger the duplicate-share rejection.
-    /// * The IRC client's own outbound flow - same shape.
-    pub async fn dag_insert(&self, events: &[Event], dag_name: &str) -> Result<Vec<blake3::Hash>> {
-        // Implementation just runs the structural-insert path;
-        // dag_insert_with_blobs reaches the same shared inner code
-        // when called with a `skip_verify=true` shortcut, which is
-        // what an empty `blobs` slice now means after the strictness
-        // tightening below - but only via this private wrapper.
+    /// This is the crate-internal post-verification entry point for callers
+    /// that have already verified the proof separately and recorded RLN
+    /// metadata. Public callers must use [`Self::insert_signal_with_blob`] or
+    /// [`Self::dag_insert_with_blobs`] so non-genesis events cannot be inserted
+    /// without their proof blob.
+    pub(crate) async fn dag_insert(
+        &self,
+        events: &[Event],
+        dag_name: &str,
+    ) -> Result<Vec<blake3::Hash>> {
         self.dag_insert_inner(events, &[], /* require_blobs */ false, dag_name).await
     }
 
+    /// Commit a rotating-DAG signal event whose RLN proof has already been
+    /// verified and recorded by the caller.
+    ///
+    /// Used by live protocol ingestion after `verify_rln_signal()` accepts the
+    /// event. The blob is persisted before the event body so late joiners never
+    /// observe a locally committed non-genesis event without its proof blob.
+    pub(crate) async fn insert_verified_signal(
+        &self,
+        event: &Event,
+        blob: &[u8],
+        dag_name: &str,
+    ) -> Result<Vec<blake3::Hash>> {
+        if event.header.parents != NULL_PARENTS && blob.is_empty() {
+            return Err(Error::Custom("verified signal event blob must not be empty".into()))
+        }
+
+        self.header_dag_insert(vec![event.header.clone()], dag_name).await?;
+        if event.header.parents != NULL_PARENTS {
+            self.dag_blob_store(&event.id(), blob)?;
+        }
+        self.dag_insert(std::slice::from_ref(event), dag_name).await
+    }
+
     /// Insert events into a rotating DAG, with mandatory RLN
     /// verification.
     ///
@@ -1957,7 +2021,8 @@ impl EventGraph {
         Ok(())
     }
 
-    pub async fn static_insert(&self, ev: &Event) -> Result<()> {
+    #[cfg(test)]
+    pub(crate) async fn static_insert(&self, ev: &Event) -> Result<()> {
         self.static_persist(ev).await?;
         self.static_pub.notify(ev.clone()).await;
         Ok(())

src/event_graph/proto.rs

@@ -24,7 +24,6 @@
 
 use std::{
     collections::{BTreeMap, HashSet, VecDeque},
-    slice,
     str::FromStr,
     sync::{
         atomic::{AtomicUsize, Ordering::SeqCst},
@@ -490,37 +489,16 @@ impl ProtocolEventGraph {
                 continue
             }
 
-            // Insert the event itself. We use plain dag_insert
-            // (not dag_insert_with_blobs) because the blob has
-            // already been verified above (the verify_rln_signal
-            // gate). Re-verifying via dag_insert_with_blobs would
-            // produce a spurious "duplicate share" rejection,
-            // because rln_verify_signal recorded the share on the
-            // first call.
-            //
-            // We store the blob in the side-table separately, so
-            // future late-joiners can re-verify it during sync.
-            // This mirrors the originator path in nickserv.rs that
-            // calls static_blob_store after static_insert.
-            if self
-                .event_graph
-                .header_dag_insert(vec![event.header.clone()], &dag_name)
-                .await
-                .is_err()
-            {
+            // Commit the already-verified signal without re-running RLN proof
+            // verification. `verify_rln_signal` above recorded the share, so
+            // the post-verification helper stores the blob, inserts the header,
+            // and then inserts the event body without calling the verifier a
+            // second time.
+            if self.event_graph.insert_verified_signal(&event, &blob, &dag_name).await.is_err() {
                 self.clone().strike().await?;
                 continue
             }
 
-            if self.event_graph.dag_insert(slice::from_ref(&event), &dag_name).await.is_err() {
-                self.clone().strike().await?;
-                continue
-            }
-
-            // Persist the verified blob alongside the event for
-            // sync-time re-verification by future late-joiners.
-            let _ = self.event_graph.dag_blob_store(&event.id(), &blob);
-
             // Relay to other peers (bounded - drops if channel full)
             let _ = self.broadcaster_push.try_send(EventPut(event, blob));
         }

src/event_graph/tests_rln.rs

@@ -1434,6 +1434,31 @@ fn rln_dag_insert_with_blobs_rejects_missing_blob_on_non_genesis() {
     })
 }
 
+#[test]
+fn rln_insert_signal_with_blob_rejects_missing_blob_on_non_genesis() {
+    // The public insertion API must not expose the internal
+    // post-verification trust path. A non-genesis signal without a
+    // blob is rejected before any header, body, or blob side-table
+    // state is persisted.
+    smol::block_on(async {
+        let eg = make_eg().await;
+        let dag_ts = eg.current_genesis.read().await.header.timestamp;
+        let dag_name = dag_ts.to_string();
+        let event = Event::new(b"public-missing-blob".to_vec(), &eg).await;
+        assert_ne!(event.header.parents, NULL_PARENTS);
+
+        let result = eg.insert_signal_with_blob(&event, &[], &dag_name).await;
+        assert!(result.is_err(), "missing blob must be rejected");
+
+        let store = eg.dag_store.read().await;
+        let slot = store.get_slot(&dag_ts).unwrap();
+        assert!(!slot.header_tree.contains_key(event.id().as_bytes()).unwrap());
+        assert!(!slot.main_tree.contains_key(event.id().as_bytes()).unwrap());
+        drop(store);
+        assert!(eg.dag_blob_fetch(&event.id()).unwrap().is_none());
+    })
+}
+
 #[test]
 fn rln_dag_insert_with_blobs_genesis_skips_verification() {
     // Genesis-shaped events (parents == NULL_PARENTS) are consensus