zodia/zodia
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

phase A wire-up: LogSync → ZodiaPipeline → AppMsg

Browse Source 
Plumbs the inbound LogSync stream through ZodiaPipeline.  The op flow
   end-to-end:

     remote peer → LogSync sub → node.inbound_ops → pipeline.process(op)
       → pipeline.next() → AppMsg::SyncStateEvent → store.insert_from_op
       → recent_interps refresh + UI bump

   Publish goes the other way: AppMsg::ShareInterp now packages the entry
   as InterpOp::Author and forwards it to node.publish(op), which CBOR-
   encodes, signs the p2panda header, persists to the operation log, and
   as InterpOp::Author and forwards it to node.publish(op), which CBOR-
   encodes, signs the p2panda header, persists to the operation log, and
   broadcasts via LogSync.

   Changes:

     * zodia-sync: drop InterpPayload (legacy Zodia-sig wrapper), drop
       ReceivedInterp, drop the ZodiaStore dep entirely.  Expose raw
       Operation<()> on `inbound_ops` and accept InterpOp on publish.
       The subscription task is now a thin forwarder — decoding lives in
       the pipeline.
     * zodia-store: add insert_from_op(key, body, author_pk) that trusts
       its caller (the pipeline) for authentication and leaves author_sig
       NULL.  Tier-1 community_for_keys still filters on
       author_sig IS NOT NULL so re-sharing is unaffected.
     * app: SyncInterpReceived → SyncStateEvent(StateEvent); the
       SyncPublishMsg::Publish payload becomes a plain InterpOp; the pump
       task moves from tokio::spawn to glib::MainContext::default()
       .spawn_local because ZodiaPipeline is !Send (single-threaded by
       design).
     * sync/Cargo.toml: zodia-store dep removed, zodia-ops dep added.

   Wire-format break: peers on this branch encode InterpOp::Author instead
   of the legacy InterpPayload struct.  Old peers' bodies will produce
   StateEvent::Skipped { MalformedOp } and be dropped silently.  That's
   acceptable for a feature branch; the release note will flag it when
   this lands.

   Workspace builds clean.  8/8 unit tests pass across zodia-ops +
   zodia-pipeline.
This commit is contained in:
Bendik Aagaard Lynghaug
2026-05-28 21:21:12 +02:00
parent 8597edb8b6
commit 36fb5c886b
6 changed files with 160 additions and 194 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+3 -1
View File
@@ -6810,6 +6810,8 @@ dependencies = [
"zodia-core",
"zodia-crypto",
"zodia-net",
"zodia-ops",
"zodia-pipeline",
"zodia-store",
"zodia-sync",
]
@@ -6942,5 +6944,5 @@ dependencies = [
"tokio",
"tracing",
"zodia-core",
"zodia-store",
"zodia-ops",
]
app/Cargo.toml
+2
View File
@@ -17,6 +17,8 @@ zodia-config.workspace = true
zodia-store.workspace = true
zodia-av.workspace = true
zodia-sync.workspace = true
zodia-ops.workspace = true
zodia-pipeline.workspace = true
p2panda-core = "0.6"
ciborium.workspace = true
blake3.workspace = true
app/src/app.rs
+71 -39
View File
@@ -30,7 +30,9 @@ use zodia_crypto::{ecies_decrypt, ecies_encrypt};
use zodia_net::{ChannelMsg, ConsentBlob, DirectChannel, InterpEntry,
NetworkConfig, PeerId, PeerStatus, RelayPayload, ZodiaNetEvent, ZodiaNetwork};
use zodia_store::{StoreError, ZodiaStore, BaselineStore};
use zodia_sync::{ReceivedInterp, ZodiaSyncNode};
use zodia_sync::ZodiaSyncNode;
use zodia_ops::InterpOp;
use zodia_pipeline::{StateEvent, ZodiaPipeline};
use relm4::factory::FactoryVecDeque;
@@ -127,8 +129,10 @@ pub enum AppMsg {
GoingOffline,
/// User submitted a new interpretation — broadcast it to all live peers.
ShareInterp(InterpEntry),
/// A new community interpretation arrived via p2panda LogSync.
SyncInterpReceived(ReceivedInterp),
/// A typed state event from the inbound `ZodiaPipeline`. Replaces
/// the legacy `SyncInterpReceived` path: now everything that arrives
/// over LogSync flows through the pipeline first.
SyncStateEvent(StateEvent),
/// User tapped the affirm button on a community interpretation row.
AffirmInterp { log_id: [u8; 32] },
/// User submitted a fresh community interpretation from a detail page.
@@ -847,27 +851,47 @@ impl AsyncComponent for AppModel {
}
// Slow path: publish to the p2panda log for offline catch-up sync.
if let Some(tx) = &self.sync_publish_tx {
if entry.author_sig.len() == 64 {
let mut sig = [0u8; 64];
sig.copy_from_slice(&entry.author_sig);
let _ = tx.try_send(SyncPublishMsg::Publish {
interp_key: entry.interp_key,
body: entry.body,
author_sig: sig,
});
}
let op = InterpOp::Author {
interp_key: entry.interp_key,
body: entry.body,
};
let _ = tx.try_send(SyncPublishMsg::Publish(op));
}
}
AppMsg::SyncInterpReceived(interp) => {
debug!(
key = %interp.interp_key,
author = %hex::encode(&interp.author_pk[..4]),
"new interpretation received via sync"
);
// Reload activity feed and trigger a network view refresh.
self.recent_interps = self.store
.recent_community_interps(12).await.unwrap_or_default();
self.network_changed_token += 1;
AppMsg::SyncStateEvent(event) => {
match event {
StateEvent::InterpAuthored { author, interp_key, body, .. } => {
let author_pk: [u8; 32] = *author.as_bytes();
match self.store
.insert_from_op(&interp_key, &body, &author_pk)
.await
{
Ok(true) => {
debug!(
key = %interp_key,
author = %hex::encode(&author_pk[..4]),
"interp authored via sync — stored"
);
self.recent_interps = self.store
.recent_community_interps(12).await.unwrap_or_default();
self.network_changed_token += 1;
}
Ok(false) => {} // duplicate, nothing to do
Err(e) => warn!("sync insert_from_op failed: {e}"),
}
}
StateEvent::AffirmAdded { .. } => {
// Phase B will wire affirmations into a store projection.
// Today: log + bump so the UI knows something changed.
self.network_changed_token += 1;
}
StateEvent::ResponseAdded { .. } => {
// Phase C will wire response threading. No-op for now.
}
StateEvent::Skipped { reason } => {
debug!(?reason, "sync op skipped");
}
}
}
AppMsg::AffirmInterp { log_id } => {
let author_pk = self.identity.public_key();
@@ -1555,14 +1579,21 @@ async fn try_spawn_network(
/// Message type for sending publish requests to the background sync task.
pub(crate) enum SyncPublishMsg {
Publish { interp_key: String, body: String, author_sig: [u8; 64] },
Publish(InterpOp),
}
/// Spawn the LogSync background task and return a channel for publishing.
/// Spawn the LogSync background pump and return a channel for publishing.
///
/// Opens a second connection to the same SQLite file so the sync task can
/// call `insert_received` without conflicting with the main-thread store
/// (WAL mode allows concurrent readers + one writer).
/// Architecture:
/// - `ZodiaSyncNode` exposes raw `Operation<()>` on `inbound_ops`.
/// - A `ZodiaPipeline` decodes / materialises each op into `StateEvent`s.
/// - We dispatch each `StateEvent` back to the model as `AppMsg::SyncStateEvent`.
/// - Outbound publishes (`SyncPublishMsg::Publish(InterpOp)`) go straight
/// to `node.publish`.
///
/// The pipeline is `!Send` (p2panda-stream is single-threaded by design),
/// so the pump task runs on glib's main-thread context via
/// `spawn_future_local`, not on tokio's multi-thread runtime.
async fn try_spawn_sync(
config: &LocalConfig,
net: &ZodiaNetwork,
@@ -1570,12 +1601,6 @@ async fn try_spawn_sync(
) -> Option<tokio::sync::mpsc::Sender<SyncPublishMsg>> {
use zodia_core::topic_key_global;
let store_path = config.data_dir().join("interpretations.db");
let sync_store = match ZodiaStore::open(&store_path).await {
Ok(s) => s,
Err(e) => { warn!("sync store open failed: {e}"); return None; }
};
let signing_key = config.identity.to_panda_key();
let topic = p2panda_core::Topic::from(topic_key_global().0);
@@ -1583,7 +1608,6 @@ async fn try_spawn_sync(
signing_key,
net.endpoint(),
net.gossip(),
sync_store,
topic,
config.data_dir(),
).await {
@@ -1594,21 +1618,29 @@ async fn try_spawn_sync(
let (tx, mut rx) = tokio::sync::mpsc::channel::<SyncPublishMsg>(32);
let sender_bg = sender.clone();
tokio::spawn(async move {
glib::MainContext::default().spawn_local(async move {
let mut node = node;
let pipeline = ZodiaPipeline::new();
loop {
tokio::select! {
Some(msg) = rx.recv() => {
match msg {
SyncPublishMsg::Publish { interp_key, body, author_sig } => {
if let Err(e) = node.publish(&interp_key, &body, &author_sig).await {
SyncPublishMsg::Publish(op) => {
if let Err(e) = node.publish(op).await {
warn!("sync publish: {e}");
}
}
}
}
Some(interp) = node.received.recv() => {
sender_bg.input(AppMsg::SyncInterpReceived(interp));
Some(op) = node.inbound_ops.recv() => {
if pipeline.process(op).await.is_err() {
warn!("pipeline closed unexpectedly");
break;
}
match pipeline.next().await {
Ok(event) => sender_bg.input(AppMsg::SyncStateEvent(event)),
Err(e) => warn!("pipeline next: {e}"),
}
}
else => break,
}
store/src/lib.rs
+38
View File
@@ -191,6 +191,44 @@ impl ZodiaStore {
Ok(log_id)
}
/// Persist an interpretation whose authentication is provided by an
/// outer container (e.g. the p2panda operation header signature from a
/// LogSync-replicated op) — no Zodia-level `author_sig` is required or
/// stored. The caller is responsible for asserting the row's
/// authenticity before invoking this.
///
/// Returns the derived `log_id`. `INSERT OR IGNORE` semantics: duplicate
/// log_ids are silently dropped.
///
/// Rows inserted this way leave the `author_sig` column NULL, which
/// means they don't participate in Tier-1 `community_for_keys` re-sharing
/// (that path filters on `author_sig IS NOT NULL`). Re-sharing happens
/// via LogSync instead, which carries the original p2panda header.
pub async fn insert_from_op(
&self,
interp_key: &str,
body: &str,
author_pk: &[u8; 32],
) -> Result<bool, StoreError> {
let log_id = derive_log_id(interp_key, body);
let kind = kind_from_key_str(interp_key);
let now = unix_secs() as i64;
let result = sqlx::query(
"INSERT OR IGNORE INTO interpretations
(log_id, interp_key, interp_kind, body, author_pk, received_at, is_baseline)
VALUES (?, ?, ?, ?, ?, ?, 0)",
)
.bind(log_id.as_slice())
.bind(interp_key)
.bind(kind)
.bind(body)
.bind(author_pk.as_slice())
.bind(now)
.execute(&self.pool)
.await?;
Ok(result.rows_affected() > 0)
}
/// Verify and insert a community interpretation received from a peer.
///
/// Returns `Ok(true)` if newly inserted, `Ok(false)` if already present,
sync/Cargo.toml
+1 -1
View File
@@ -7,7 +7,7 @@ description = "Offline-first interpretation sync built on p2panda LogSync"
[dependencies]
zodia-core.workspace = true
zodia-store.workspace = true
zodia-ops.workspace = true
serde.workspace = true
thiserror.workspace = true
tokio.workspace = true
sync/src/lib.rs
+45 -153
View File
@@ -1,15 +1,19 @@
//! Offline-first interpretation sync for the Zodia community index.
//! Offline-first interpretation sync driver.
//!
//! # Design
//! `ZodiaSyncNode` owns the `p2panda-net::LogSync` machinery and exposes
//! two simple channels to the app layer:
//!
//! Each user maintains an append-only p2panda log (log id `0`) of the
//! interpretations they have authored. When two peers share the same sync
//! topic they perform a **set-reconciliation catch-up** (exchanging log
//! heights) and then enter **live mode** where newly published operations
//! are gossip-broadcast immediately.
//! * `inbound_ops` — every received `Operation<()>`, raw. The app feeds
//! these into a `zodia-pipeline::ZodiaPipeline` for decoding, ordering,
//! access-control, materialisation.
//! * `publish(op: InterpOp)` — encode the canonical Zodia op into a body,
//! build + sign the p2panda header, persist to the local log, broadcast
//! via LogSync.
//!
//! `ZodiaSyncNode` wraps `p2panda-net`'s `LogSync` and translates between the
//! p2panda operation layer and `ZodiaStore`'s application-level records.
//! The Zodia-level "author_sig over BLAKE3(key||body)" that the old
//! `InterpPayload` carried is gone: the p2panda header signature IS the
//! authentication for LogSync-replicated ops, and `zodia-store::insert_from_op`
//! trusts that the caller (the pipeline) verified the chain.
//!
//! # Storage
//!
@@ -28,76 +32,31 @@ use p2panda_store::logs::LogStore;
use p2panda_store::operations::OperationStore;
use p2panda_store::{SqliteStore, SqliteStoreBuilder};
use p2panda_sync::protocols::TopicLogSyncEvent;
use serde::{Deserialize, Serialize};
use thiserror::Error;
use tokio::sync::mpsc;
use tracing::{debug, warn};
use zodia_store::{StoreError, ZodiaStore};
use zodia_ops::InterpOp;
// ── log id ────────────────────────────────────────────────────────────────────
/// Each author has exactly one log containing all their interpretations.
const INTERP_LOG_ID: u64 = 0;
// ── payload ───────────────────────────────────────────────────────────────────
// ── errors ────────────────────────────────────────────────────────────────────
/// CBOR-encoded body of a p2panda interpretation operation.
///
/// The `author_sig` is the Zodia-level ed25519 signature over
/// `BLAKE3(interp_key || body)` — the same payload verified by
/// `ZodiaStore::insert_received`. Because the p2panda header already carries
/// an ed25519 signature from the same key pair, this is redundant but lets us
/// feed received operations directly into `ZodiaStore::insert_received`
/// without modifying its verification contract.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct InterpPayload {
pub interp_key: String,
pub body: String,
/// ed25519 signature, 64 bytes.
pub author_sig: Vec<u8>,
}
impl InterpPayload {
fn encode(&self) -> Vec<u8> {
let mut buf = Vec::new();
ciborium::into_writer(self, &mut buf).expect("ciborium encode infallible");
buf
}
fn decode(bytes: &[u8]) -> Option<Self> {
ciborium::from_reader(bytes).ok()
}
}
// ── sync node ─────────────────────────────────────────────────────────────────
/// Error type for sync operations.
#[derive(Debug, Error)]
pub enum SyncError {
#[error("store error: {0}")]
Store(#[from] StoreError),
#[error("p2panda store: {0}")]
PandaStore(String),
#[error("p2panda sync: {0}")]
Sync(String),
#[error("payload encode/decode failed")]
Payload,
}
/// A decoded interpretation that arrived via LogSync from a remote peer.
#[derive(Debug, Clone)]
pub struct ReceivedInterp {
pub interp_key: String,
pub body: String,
pub author_pk: [u8; 32],
pub author_sig: [u8; 64],
}
// ── sync node ─────────────────────────────────────────────────────────────────
/// The live sync handle.
///
/// Keeps the p2panda LogSync machinery alive and mediates between the
/// application and the sync layer.
/// The live sync handle. Keeps the p2panda LogSync machinery alive and
/// exposes a raw `Operation<()>` channel for the app's pipeline to consume.
pub struct ZodiaSyncNode {
/// p2panda signing key — same bytes as the Zodia identity `SigningKey`.
signing_key: SigningKey,
@@ -105,8 +64,9 @@ pub struct ZodiaSyncNode {
sync_store: SqliteStore,
/// LogSync handle for our single sync topic.
handle: p2panda_net::sync::SyncHandle<Operation<()>, TopicLogSyncEvent<()>>,
/// Interpretations received from remote peers, ready for the app to consume.
pub received: mpsc::Receiver<ReceivedInterp>,
/// Raw operations received from remote peers, ready for the app's
/// `ZodiaPipeline` to consume.
pub inbound_ops: mpsc::Receiver<Operation<()>>,
}
impl ZodiaSyncNode {
@@ -115,15 +75,12 @@ impl ZodiaSyncNode {
/// * `signing_key` — the local identity p2panda `SigningKey`
/// * `endpoint` — clone of `ZodiaNetwork`'s iroh endpoint
/// * `gossip` — clone of `ZodiaNetwork`'s gossip engine
/// * `zodia_store` — shared handle to the application `ZodiaStore` for
/// persisting received interpretations
/// * `sync_topic` — the sync topic (use `Topic::from(topic_key_global().0)`)
/// * `store_dir` — directory in which the sync-store SQLite file lives
pub async fn spawn(
signing_key: SigningKey,
endpoint: Endpoint,
gossip: Gossip,
zodia_store: ZodiaStore,
sync_topic: Topic,
store_dir: &Path,
) -> Result<Self, SyncError> {
@@ -147,16 +104,19 @@ impl ZodiaSyncNode {
.await
.map_err(|e| SyncError::Sync(format!("{e:?}")))?;
let (recv_tx, recv_rx) = mpsc::channel(256);
let (op_tx, op_rx) = mpsc::channel::<Operation<()>>(256);
// ── subscription background task ──────────────────────────────────────
//
// The task is intentionally thin: forward every received operation
// to the channel and log non-operation lifecycle events. All
// decoding / verification / storage decisions happen downstream in
// the app's `ZodiaPipeline`.
let mut subscription = handle
.subscribe()
.await
.map_err(|e| SyncError::Sync(format!("{e:?}")))?;
let zodia_store_bg = zodia_store.clone();
tokio::spawn(async move {
while let Some(result) = subscription.next().await {
let from_sync = match result {
@@ -167,76 +127,24 @@ impl ZodiaSyncNode {
}
};
let remote_tag = hex::encode(&from_sync.remote.as_bytes()[..4]);
match from_sync.event {
TopicLogSyncEvent::OperationReceived { operation, .. } => {
let author_pk_bytes: [u8; 32] =
*operation.header.verifying_key.as_bytes();
let body_bytes = match &operation.body {
Some(b) => b.to_bytes(),
None => {
debug!("sync: operation without body, skipping");
continue;
}
};
let payload = match InterpPayload::decode(&body_bytes) {
Some(p) => p,
None => {
warn!("sync: failed to decode InterpPayload");
continue;
}
};
if payload.author_sig.len() != 64 {
warn!("sync: author_sig wrong length ({})", payload.author_sig.len());
continue;
}
let mut sig_arr = [0u8; 64];
sig_arr.copy_from_slice(&payload.author_sig);
let interp_key = payload.interp_key.clone();
let body_text = payload.body.clone();
match zodia_store_bg
.insert_received(&interp_key, &body_text, &author_pk_bytes, &sig_arr)
.await
{
Ok(true) => {
debug!(key = %payload.interp_key, "sync: new interpretation received");
let _ = recv_tx.send(ReceivedInterp {
interp_key: payload.interp_key,
body: payload.body,
author_pk: author_pk_bytes,
author_sig: sig_arr,
}).await;
}
Ok(false) => {
debug!(key = %payload.interp_key, "sync: duplicate, skipped");
}
Err(StoreError::InvalidSignature) => {
warn!(key = %payload.interp_key, "sync: invalid sig, discarded");
}
Err(e) => {
warn!(key = %payload.interp_key, "sync: store error: {e}");
}
// `Box<Operation<()>>` → owned `Operation<()>`.
let op = *operation;
if op_tx.send(op).await.is_err() {
debug!("inbound_ops channel closed, stopping subscription pump");
break;
}
}
TopicLogSyncEvent::SyncStarted { .. } => {
debug!(
remote = %hex::encode(&from_sync.remote.as_bytes()[..4]),
"sync: catch-up started"
);
debug!(remote = %remote_tag, "sync: catch-up started");
}
TopicLogSyncEvent::SyncFinished { .. } => {
debug!(
remote = %hex::encode(&from_sync.remote.as_bytes()[..4]),
"sync: catch-up finished"
);
debug!(remote = %remote_tag, "sync: catch-up finished");
}
TopicLogSyncEvent::Failed { error } => {
warn!(remote = %hex::encode(&from_sync.remote.as_bytes()[..4]),
"sync session failed: {error}");
warn!(remote = %remote_tag, "sync session failed: {error}");
}
_ => {}
}
@@ -247,30 +155,17 @@ impl ZodiaSyncNode {
signing_key,
sync_store,
handle,
received: recv_rx,
inbound_ops: op_rx,
})
}
/// Publish a locally authored interpretation to the p2panda log.
/// Publish a locally authored `InterpOp` to the p2panda log.
///
/// Callers must have already inserted the entry into `ZodiaStore` via
/// `insert_signed`. This method adds the operation to the p2panda log
/// so that it will be propagated to peers via gossip and catch-up sync.
///
/// `author_sig` is the Zodia-level ed25519 signature already stored in
/// `ZodiaStore` — 64 bytes.
pub async fn publish(
&mut self,
interp_key: &str,
body: &str,
author_sig: &[u8; 64],
) -> Result<(), SyncError> {
let payload = InterpPayload {
interp_key: interp_key.to_owned(),
body: body.to_owned(),
author_sig: author_sig.to_vec(),
};
let payload_bytes = payload.encode();
/// Encodes the op, builds + signs a p2panda header, persists locally
/// (so the next publish picks up the right backlink and crash-mid-publish
/// recovers), then broadcasts via LogSync.
pub async fn publish(&mut self, op: InterpOp) -> Result<(), SyncError> {
let payload_bytes = op.encode();
// Determine the next sequence number + backlink from our log tip.
let latest: Option<Operation<()>> = self.sync_store
@@ -279,8 +174,8 @@ impl ZodiaSyncNode {
.map_err(|e| SyncError::PandaStore(e.to_string()))?;
let (seq_num, backlink) = match latest {
Some(op) => (op.header.seq_num + 1, Some(op.header.hash())),
None => (0, None),
Some(prev) => (prev.header.seq_num + 1, Some(prev.header.hash())),
None => (0, None),
};
let body_op = Body::new(&payload_bytes);
@@ -305,14 +200,11 @@ impl ZodiaSyncNode {
body: Some(body_op),
};
// Persist locally so the next publish picks up the right backlink and
// peers that catch us mid-publish can complete the log.
self.sync_store
.insert_operation(&op_hash, &operation, &INTERP_LOG_ID)
.await
.map_err(|e| SyncError::PandaStore(e.to_string()))?;
// Broadcast to connected peers via gossip.
self.handle
.publish(operation)
.await