Allow for our moderation service to be run on multiple different servers at the same time

Author: markharding

services/common/kafka/src/lib.rs

@@ -5,6 +5,7 @@ mod config;
 mod consumer;
 mod producer;
 
+pub use rdkafka::Offset;
 pub use rdkafka::message::{BorrowedMessage, Message};
 
 pub use consumer::{CommitMode, Consumer, CustomContext, build_consumer};

services/moderation/src/main.rs

@@ -7,10 +7,14 @@ mod polycentric;
 mod providers;
 mod repository;
 
-use common_kafka::{BorrowedMessage, CommitMode, Consumer, Message};
+use std::collections::HashMap;
+use std::time::Duration;
+
+use common_kafka::{BorrowedMessage, CommitMode, Consumer, Message, Offset};
 use common_object_store::{ObjectStore, ObjectStoreConfig};
 use context::Context;
 use polycentric::{PolycentricClient, PublishError};
+use polycentric_common::models::collections;
 use polycentric_common::models::protos_v2::{
     Content, ContentDigest, Event, EventBundle, EventKey, ImageSet, content::ContentBody,
 };
@@ -20,11 +24,19 @@ use providers::azure::{AzureClient, ModerationRequest};
 use prost::Message as ProstMessage;
 use sea_orm::sea_query::value::prelude::serde_json;
 
+/// Duration before retrying a Retry event
+const RETRY_BACKOFF: Duration = Duration::from_secs(2);
+
+/// Number of times a message is retried before it is skipped (committed
+/// past without being processed).
+const MAX_RETRIES: u32 = 5;
+
 /// Whether the consumed message's offset should be committed.
 enum Outcome {
     /// Done with this message — commit so it is not redelivered.
     Commit,
-    /// Transient failure — leave uncommitted so it is retried.
+    /// Transient failure — seek back so the message is re-delivered and
+    /// retried (see the consume loop). After [`MAX_RETRIES`] it is skipped.
     Retry,
 }
 
@@ -221,6 +233,9 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // Listen to a Kafka topic of all events.
     let consumer = common_kafka::build_consumer("events", &["events"]).await;
 
+    // Failure counts for messages currently being retried.
+    let mut attempts: HashMap<(i32, i64), u32> = HashMap::new();
+
     loop {
         let message = match consumer.recv().await {
             Ok(message) => message,
@@ -230,13 +245,47 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             }
         };
 
+        let coord = (message.partition(), message.offset());
+
         match process(&ctx, &message).await {
             Outcome::Commit => {
+                attempts.remove(&coord);
                 if let Err(e) = consumer.commit_message(&message, CommitMode::Async) {
                     warn!("failed to commit offset: {}", e);
                 }
             }
-            Outcome::Retry => {}
+            Outcome::Retry => {
+                let failures = {
+                    let count = attempts.entry(coord).or_insert(0);
+                    *count += 1;
+                    *count
+                };
+
+                if failures > MAX_RETRIES {
+                    // Retries exhausted — give up and commit past this message
+                    // so the partition can make progress.
+                    warn!(
+                        "message at partition {} offset {} failed {} times; skipping",
+                        coord.0, coord.1, failures
+                    );
+                    attempts.remove(&coord);
+                    if let Err(e) = consumer.commit_message(&message, CommitMode::Async) {
+                        warn!("failed to commit offset after skip: {}", e);
+                    }
+                } else {
+                    // Seek back so the next poll re-delivers this message, then
+                    // back off to avoid a hot loop.
+                    if let Err(e) = consumer.seek(
+                        message.topic(),
+                        message.partition(),
+                        Offset::Offset(message.offset()),
+                        Duration::from_secs(5),
+                    ) {
+                        warn!("failed to seek for retry: {}", e);
+                    }
+                    tokio::time::sleep(RETRY_BACKOFF).await;
+                }
+            }
         }
     }
 }
@@ -376,7 +425,24 @@ async fn process(ctx: &Context, message: &BorrowedMessage<'_>) -> Outcome {
 /// moderation DB. Returns `Retry` on any failure (transient, persistence,
 /// or a not-ready identity) so the label is never silently dropped.
 async fn publish_labels(ctx: &Context, target: EventKey, labels: Vec<String>) -> Outcome {
-    match ctx.polycentric.publish_labels(target, labels).await {
+    let signer = ctx.polycentric.public_key();
+    let head = match repository::chain_head(
+        &ctx.db,
+        collections::LABELS,
+        ctx.polycentric.identity(),
+        signer.key_type,
+        &signer.key,
+    )
+    .await
+    {
+        Ok(head) => head,
+        Err(e) => {
+            warn!("chain_head error: {}", e);
+            return Outcome::Retry;
+        }
+    };
+
+    match ctx.polycentric.publish_labels(target, labels, head).await {
         Ok(created) => match repository::persist_created(&ctx.db, &created).await {
             Ok(()) => Outcome::Commit,
             Err(e) => {

services/moderation/src/polycentric.rs

@@ -34,6 +34,18 @@ impl std::fmt::Display for PublishError {
     }
 }
 
+/// Chain position for the next event we author in a collection, read from
+/// the durable Postgres store (not local in-memory state) so that multiple
+/// moderation processes extend a single, shared chain consistently.
+pub struct ChainHead {
+    /// Sequence to assign the next event (`max(sequence) + 1`, or 1).
+    pub next_sequence: u64,
+    /// Canonically-latest signature in the collection (empty if none).
+    pub previous_signature: Vec<u8>,
+    /// Merkle root over the collection's canonical signatures (empty if none).
+    pub previous_root: Vec<u8>,
+}
+
 /// Signed event and content that should persist to the database.
 pub struct CreatedEvent {
     /// The decoded event (carries key, digest, previous_root, etc.).
@@ -155,13 +167,22 @@ impl PolycentricClient {
         Ok(count)
     }
 
-    /// Build, sign, and push a `Labels` event for `target`, then record it
-    /// locally so the next event chains correctly. Returns the created
-    /// event for persistence by the caller.
+    /// The hex identity string this service publishes under.
+    pub fn identity(&self) -> &str {
+        &self.identity
+    }
+
+    /// The public key this service signs events with.
+    pub fn public_key(&self) -> &PublicKey {
+        &self.public_key
+    }
+
+    /// Build, sign, and push a `Labels` event for `target`.
     pub async fn publish_labels(
         &self,
         target: EventKey,
         label_values: Vec<String>,
+        head: ChainHead,
     ) -> Result<CreatedEvent, PublishError> {
         let identity_sequence = self.identity_sequence.load(Ordering::Relaxed);
         if identity_sequence == 0 {
@@ -172,38 +193,36 @@ impl PolycentricClient {
 
         let (content_bytes, digest) = labels_content(&target, &label_values);
 
-        // Compute the chain position synchronously, without holding the lock
-        // across any await.
-        let event = {
+        // The sequence and prior-chain references come from the durable
+        // Postgres store (`head`); only the vector clock is derived locally,
+        // from the static identity chain loaded at bootstrap.
+        let sequence = head.next_sequence;
+        let vector_clock = {
             let core = self.core.lock().unwrap();
-            let sequence = core.next_sequence(&self.identity, collections::LABELS);
-            let vector_clock = core
-                .build_vector_clock(
-                    &self.identity,
-                    collections::LABELS,
-                    identity_sequence,
-                    &self.public_key,
-                    sequence,
-                    None,
-                )
-                .map_err(|e| PublishError::NotReady(e.to_string()))?;
-            let previous_root = core.previous_root(&self.identity, collections::LABELS);
-            let previous_signature = core.previous_signature(&self.identity, collections::LABELS);
-
-            Event {
-                key: Some(EventKey {
-                    collection: collections::LABELS,
-                    identity: self.identity.clone(),
-                    signed_by: Some(self.public_key.clone()),
-                    sequence,
-                }),
+            core.build_vector_clock(
+                &self.identity,
+                collections::LABELS,
                 identity_sequence,
-                vector_clock: Some(vector_clock),
-                previous_signature,
-                previous_root,
-                content_digest: Some(digest),
-                created_at: now_millis(),
-            }
+                &self.public_key,
+                sequence,
+                None,
+            )
+            .map_err(|e| PublishError::NotReady(e.to_string()))?
+        };
+
+        let event = Event {
+            key: Some(EventKey {
+                collection: collections::LABELS,
+                identity: self.identity.clone(),
+                signed_by: Some(self.public_key.clone()),
+                sequence,
+            }),
+            identity_sequence,
+            vector_clock: Some(vector_clock),
+            previous_signature: head.previous_signature,
+            previous_root: head.previous_root,
+            content_digest: Some(digest),
+            created_at: now_millis(),
         };
 
         let event_bytes = event.encode_to_vec();
@@ -223,7 +242,7 @@ impl PolycentricClient {
         // Push to every server; success on any is enough to consider it
         // published (servers gossip among themselves).
         let request = PutEventsRequest {
-            event_bundles: vec![bundle.clone()],
+            event_bundles: vec![bundle],
         };
         let mut pushed = false;
         for server in &self.servers {
@@ -238,9 +257,8 @@ impl PolycentricClient {
             ));
         }
 
-        // Record locally so subsequent events extend this chain.
-        self.core.lock().unwrap().copy_bundles(vec![bundle]);
-
+        // No local state to update: the next event's chain position is read
+        // back from Postgres once the caller persists this one.
         Ok(CreatedEvent {
             event,
             signature,

services/moderation/src/repository.rs

@@ -2,14 +2,15 @@ use moderation_entity::processed_content_model::{
     ActiveModel, Entity as ProcessedContent, Model as ProcessedContentModel, Status,
 };
 use moderation_entity::{created_content_model, created_event_model};
+use polycentric_common::merkle;
 use sea_orm::{
-    ActiveModelTrait, ActiveValue::NotSet, ActiveValue::Set, ConnectionTrait, DatabaseConnection,
-    DbErr, EntityTrait, TransactionTrait, prelude::TimeDateTime, sea_query::OnConflict,
-    sea_query::value::prelude::serde_json,
+    ActiveModelTrait, ActiveValue::NotSet, ActiveValue::Set, ColumnTrait, ConnectionTrait,
+    DatabaseConnection, DbErr, EntityTrait, QueryFilter, TransactionTrait, prelude::TimeDateTime,
+    sea_query::OnConflict, sea_query::value::prelude::serde_json,
 };
 use time::OffsetDateTime;
 
-use crate::polycentric::CreatedEvent;
+use crate::polycentric::{ChainHead, CreatedEvent};
 
 /// Current wall-clock time as the `TimeDateTime` (naive) the schema uses.
 fn now() -> TimeDateTime {
@@ -87,6 +88,47 @@ pub async fn persist_created(db: &DatabaseConnection, created: &CreatedEvent) ->
     txn.commit().await
 }
 
+/// Read the next chain position for events we author in
+pub async fn chain_head<C: ConnectionTrait>(
+    db: &C,
+    collection: i32,
+    identity: &str,
+    public_key_type: i32,
+    public_key: &[u8],
+) -> Result<ChainHead, DbErr> {
+    let rows = created_event_model::Entity::find()
+        .filter(created_event_model::Column::Collection.eq(collection))
+        .filter(created_event_model::Column::Identity.eq(identity))
+        .filter(created_event_model::Column::PublicKeyType.eq(public_key_type))
+        .filter(created_event_model::Column::PublicKey.eq(public_key.to_vec()))
+        .all(db)
+        .await?;
+
+    let next_sequence = rows
+        .iter()
+        .map(|r| r.sequence)
+        .max()
+        .map(|s| s as u64 + 1)
+        .unwrap_or(1);
+
+    // Canonical ordering depends on each event's full bytes (vector-clock
+    // sum, created_at) paired with its signature — both stored per row.
+    let canonical = merkle::canonical_signatures(
+        rows.iter()
+            .map(|r| (r.event_bytes.as_slice(), r.signature.as_slice())),
+    );
+    let previous_signature = canonical.last().cloned().unwrap_or_default();
+    let previous_root = merkle::merkle_tree_hash(&canonical)
+        .map(|h| h.to_vec())
+        .unwrap_or_default();
+
+    Ok(ChainHead {
+        next_sequence,
+        previous_signature,
+        previous_root,
+    })
+}
+
 /// Returns if already stored content with this digest
 pub async fn created_content_exists<C: ConnectionTrait>(
     db: &C,