feat: link journal

Cargo.toml

@@ -32,7 +32,7 @@ alloy-primitives = "1.5"
 alloy-signer = "1.1"
 alloy-signer-local = "1.1"
 auto_impl = "1.3"
-axum = "0.8"
+axum = { version = "0.8", default-features = false }
 axum-extra = "0.12"
 bump-scope = { version = "1.5", features = ["nightly"] }
 bytes = "1.11"

crates/calendar/Cargo.toml

@@ -12,16 +12,22 @@ version.workspace = true
 alloy-primitives = { workspace = true }
 alloy-signer = { workspace = true }
 alloy-signer-local = { workspace = true }
-axum = { workspace = true, features = ["macros"] }
+axum = { workspace = true, default-features = false, features = [
+  "macros",
+  "http2",
+  "tokio",
+] } # http2 only
 bump-scope.workspace = true
 bytes = { workspace = true }
+digest = { workspace = true }
 eyre = { workspace = true }
 itoa = { workspace = true }
 sha3 = { workspace = true }
 tokio = { workspace = true, features = ["full"] }
 tracing = { workspace = true }
 tracing-subscriber = { workspace = true }
 uts-core = { workspace = true, features = ["bytes"] }
+uts-journal = { workspace = true }
 
 [dev-dependencies]
 criterion.workspace = true
@@ -34,4 +40,5 @@ name = "submit_digest"
 workspace = true
 
 [features]
+dev = ["axum/http1"] # for easier testing
 performance = ["tracing/release_max_level_info"]

crates/calendar/src/lib.rs

@@ -8,6 +8,9 @@ extern crate tracing;
 
 use alloy_signer::k256::ecdsa::SigningKey;
 use alloy_signer_local::LocalSigner;
+use digest::{OutputSizeUser, typenum::Unsigned};
+use sha3::Keccak256;
+use uts_journal::Journal;
 
 /// Calendar server routes and handlers.
 pub mod routes;
@@ -19,6 +22,8 @@ pub mod time;
 pub struct AppState {
     /// Local signer for signing OTS timestamps.
     pub signer: LocalSigner<SigningKey>,
+    /// Journal
+    pub journal: Journal<{ <Keccak256 as OutputSizeUser>::OutputSize::USIZE }>,
 }
 
 /// Signal for graceful shutdown.

crates/calendar/src/main.rs

@@ -9,6 +9,9 @@ use axum::{
 };
 use std::sync::Arc;
 use uts_calendar::{AppState, routes, shutdown_signal, time};
+use uts_journal::Journal;
+
+const RING_BUFFER_CAPACITY: usize = 1 << 20; // 1 million entries
 
 #[tokio::main]
 async fn main() -> eyre::Result<()> {
@@ -19,6 +22,10 @@ async fn main() -> eyre::Result<()> {
     let signer = LocalSigner::from_bytes(&b256!(
         "9ba9926331eb5f4995f1e358f57ba1faab8b005b51928d2fdaea16e69a6ad225"
     ))?;
+    let journal = Journal::with_capacity(RING_BUFFER_CAPACITY);
+
+    let _reader = journal.reader();
+    // TODO: spawn stamper task
 
     let app = Router::new()
         .route(
@@ -30,7 +37,7 @@ async fn main() -> eyre::Result<()> {
             "/timestamp/{hex_commitment}",
             get(routes::ots::get_timestamp),
         )
-        .with_state(Arc::new(AppState { signer }));
+        .with_state(Arc::new(AppState { signer, journal }));
 
     let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await?;
 

crates/calendar/src/routes/ots.rs

@@ -1,9 +1,15 @@
 use crate::{AppState, time::current_time_sec};
 use alloy_signer::SignerSync;
-use axum::{body::Bytes, extract::State};
+use axum::{
+    body::Bytes,
+    extract::State,
+    http::StatusCode,
+    response::{IntoResponse, Response},
+};
 use bump_scope::Bump;
 use bytes::BytesMut;
-use sha3::{Digest, Keccak256};
+use digest::Digest;
+use sha3::Keccak256;
 use std::{cell::RefCell, sync::Arc};
 use uts_core::{
     codec::{
@@ -36,10 +42,15 @@ pub const MAX_DIGEST_SIZE: usize = 64; // e.g., SHA3-512
 // result attested by Pending: update URI https://localhost:3000
 // ```
 /// Submit digest to calendar server and get pending timestamp in response.
-pub async fn submit_digest(State(state): State<Arc<AppState>>, digest: Bytes) -> Bytes {
-    let (output, _commitment) = submit_digest_inner(digest, &state.signer);
-    // TODO: submit commitment to journal
-    output
+pub async fn submit_digest(State(state): State<Arc<AppState>>, digest: Bytes) -> Response {
+    let (output, commitment) = submit_digest_inner(digest, &state.signer);
+    match state.journal.try_commit(&commitment) {
+        Err(_) => {
+            return (StatusCode::SERVICE_UNAVAILABLE, r#"{"err":"server busy"}"#).into_response();
+        } // journal is full
+        Ok(fut) => fut.await,
+    }
+    output.into_response()
 }
 
 // TODO: We need to benchmark this.
@@ -114,7 +125,7 @@ pub fn submit_digest_inner(digest: Bytes, signer: impl SignerSync) -> (Bytes, [u
         timestamp.encode(&mut buf).unwrap();
 
         #[cfg(any(debug_assertions, not(feature = "performance")))]
-        trace!(timestamp = ?timestamp, encoded_length = buf.len());
+        trace!(encoded_length = buf.len(), timestamp = ?timestamp);
 
         (buf.freeze(), commitment)
     })

crates/core/src/codec/v1/attestation.rs

@@ -27,14 +27,23 @@ const PENDING_TAG: &[u8; 8] = b"\x83\xdf\xe3\x0d\x2e\xf9\x0c\x8e";
 pub type AttestationTag = [u8; TAG_SIZE];
 
 /// Raw Proof that some data existed at a given time.
-#[derive(Clone, Debug)]
+#[derive(Clone)]
 pub struct RawAttestation<A: Allocator = Global> {
     pub tag: AttestationTag,
     pub data: Vec<u8, A>,
     /// Cached value for verifying the attestation.
     pub(crate) value: OnceLock<Vec<u8, A>>,
 }
 
+impl<A: Allocator> fmt::Debug for RawAttestation<A> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("RawAttestation")
+            .field("tag", &Hexed(&self.tag))
+            .field("data", &Hexed(&self.data))
+            .finish()
+    }
+}
+
 impl<A: Allocator> DecodeIn<A> for RawAttestation<A> {
     fn decode_in(decoder: &mut impl Decoder, alloc: A) -> Result<Self, DecodeError> {
         let mut tag = [0u8; TAG_SIZE];

crates/journal/src/lib.rs

@@ -227,7 +227,7 @@ pub struct CommitFuture<'a, const ENTRY_SIZE: usize> {
     active_waker: Option<Waker>,
 }
 
-impl<'a, const ENTRY_SIZE: usize> Future for CommitFuture<'a, ENTRY_SIZE> {
+impl<const ENTRY_SIZE: usize> Future for CommitFuture<'_, ENTRY_SIZE> {
     type Output = ();
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<Self::Output> {

crates/journal/src/reader.rs

@@ -70,7 +70,7 @@ impl<const ENTRY_SIZE: usize> JournalReader<ENTRY_SIZE> {
             target_index: u64,
         }
 
-        impl<'a, const ENTRY_SIZE: usize> Future for WaitForBatch<'a, ENTRY_SIZE> {
+        impl<const ENTRY_SIZE: usize> Future for WaitForBatch<'_, ENTRY_SIZE> {
             type Output = ();
             fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
                 if self.reader.journal.persisted_index.load(Ordering::Acquire) >= self.target_index