Update sigma-stream

.github/ci.yml → .github/workflows/ci.yml

@@ -1,20 +1,19 @@
 name: ci
-
 on:
   push:
     branches:
       - main
   pull_request:
-
 concurrency:
   group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
   cancel-in-progress: true
-
 jobs:
   cargo:
     runs-on: ubuntu-latest
     env:
       RUSTFLAGS: "-D warnings"
+      CARGO_NET_GIT_FETCH_WITH_CLI: "true"
+      CARGO_REGISTRIES_CRATES_IO_PROTOCOL: "sparse"
     strategy:
       fail-fast: false
       matrix:
@@ -27,15 +26,31 @@ jobs:
           - command: test --all-features --locked --all
           - command: doc --all-features --locked --no-deps
           - command: bench --no-run --locked --all
-
     steps:
+      - uses: webfactory/ssh-agent@v0.9.0
+        with:
+          ssh-private-key: ${{ secrets.SIGMA_NODE_SSH_KEY }}
       - uses: actions/checkout@v3
       - uses: dtolnay/rust-toolchain@stable
         with:
           targets: wasm32-unknown-unknown
       - uses: Swatinem/rust-cache@v2
-      - run: cargo ${{ matrix.command }}
-
+
+      # For test commands, we need sigma-node available
+      - name: Build sigma-node
+        if: contains(matrix.command, 'test')
+        run: |
+          # Clone and build sigma-node directly
+          git clone git@github.com:essential-contributions/sigma-node.git
+          cd sigma-node
+          cargo build --release --bin sigma-node
+          # Add to PATH
+          echo "$PWD/target/release" >> $GITHUB_PATH
+          cd ..
+
+      - name: Run cargo command
+        run: cargo ${{ matrix.command }}
+
   cargo-toml-lint:
     runs-on: ubuntu-latest
     steps:
@@ -44,3 +59,13 @@ jobs:
       - uses: Swatinem/rust-cache@v2
       - run: cargo install --version "0.1.1" cargo-toml-lint
       - run: git ls-files | grep Cargo.toml$ | xargs --verbose -n 1 cargo-toml-lint
+
+  nix:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - uses: webfactory/ssh-agent@v0.9.0
+        with:
+          ssh-private-key: ${{ secrets.SIGMA_NODE_SSH_KEY }}
+      - uses: DeterminateSystems/nix-installer-action@v10
+      - run: nix flake check --print-build-logs

Cargo.toml

@@ -15,5 +15,6 @@ futures = "0.3.31"
 reqwest = "0.12.15"
 serde = "1.0.219"
 serde_json = "1.0.140"
-sigma-types = { git = "ssh://git@github.com/essential-contributions/sigma-node.git" }
-tokio-util = "0.7.15"
+sigma-types = { git = "ssh://git@github.com/essential-contributions/sigma-node.git"}
+tokio = { version = "1.44.0", features = ["full"] }
+tokio-util = "0.7.15"

config/test.yaml

@@ -0,0 +1,8 @@
+# Test configuration for sigma-node
+# This configuration is used for integration tests only
+
+hash_type: Sha256
+signature_type: !EcdsaTestingKey
+stream_config:
+- !SysTime
+  interval_ms: 1000

crates/sigma-stream/Cargo.toml

@@ -14,4 +14,5 @@ reqwest = { workspace = true, features = ["json", "stream", "native-tls-alpn"] }
 serde.workspace = true
 serde_json.workspace = true
 sigma-types.workspace = true
-tokio-util.workspace = true
+tokio.workspace = true
+tokio-util.workspace = true

crates/sigma-stream/src/lib.rs

@@ -1,42 +1,42 @@
 #![deny(missing_docs)]
 //! Sigma Stream is a library for streaming events from a Sigma node.
-
 use futures::{Stream, TryStreamExt};
 use reqwest::{ClientBuilder, Url};
-use sigma_types::Event;
+use sigma_types::{Event, Signed};
 use std::marker::PhantomData;
 use tokio_util::{
     bytes::{self, Buf},
     codec::{Decoder, FramedRead},
     io::StreamReader,
 };
 
-/// The height of an event
-pub type EventHeight = u64;
+#[cfg(test)]
+mod tests;
 
-/// The output type of the event stream.
-pub type Output = (EventHeight, Event);
+/// The output type of the event stream - a signed event from the node.
+pub type Output = Signed<Event>;
 
 /// Create the stream of events from the node endpoint.
 pub async fn stream_events(
     url: &Url,
-    domain: &alloy::primitives::Address,
     start_event_height: usize,
-) -> Result<impl Stream<Item = Result<(EventHeight, Event), std::io::Error>>, std::io::Error> {
+) -> Result<impl Stream<Item = Result<Signed<Event>, std::io::Error>>, std::io::Error> {
     let client = ClientBuilder::new()
         .http2_prior_knowledge()
         .build()
         .map_err(std::io::Error::other)?;
+
     // Create the subscription to the node's batch stream.
     let url = url
-        .join(&format!(
-            "/subscribe-window/eth:{}/{}",
-            domain, start_event_height
-        ))
+        .join(&format!("/subscribe-events/{start_event_height}"))
         .unwrap();
 
     // Send the request to the node.
-    let response = client.get(url).send().await.unwrap();
+    let response = client
+        .get(url)
+        .send()
+        .await
+        .map_err(|e| std::io::Error::other(format!("Request failed: {e}")))?;
 
     // Check if the node returned a bad response.
     if !response.status().is_success() {
@@ -55,8 +55,7 @@ pub async fn stream_events(
     let stream = StreamReader::new(response.bytes_stream().map_err(std::io::Error::other));
 
     // Decode the stream from the node.
-    let stream = FramedRead::new(stream, SseDecoder::<(EventHeight, Event)>::new());
-
+    let stream = FramedRead::new(stream, SseDecoder::<Signed<Event>>::new());
     Ok(stream)
 }
 
@@ -89,7 +88,6 @@ where
                 let Ok(s) = std::str::from_utf8(&buf[..end]) else {
                     // If this fails we still have to advance the buffer.
                     buf.advance(end + 2);
-
                     // This will skip this bad data.
                     return Ok(None);
                 };
@@ -99,7 +97,6 @@ where
 
                 // Parse the data from the stream.
                 let data = serde_json::from_str::<T>(s);
-
                 let r = match data {
                     // Success data found.
                     Ok(data) => Ok(Some(data)),
@@ -110,7 +107,7 @@ where
                             Ok(None)
                         } else {
                             // This is a stream error.
-                            Err(std::io::Error::other(format!("Stream: {}. Error {}", s, e)))
+                            Err(std::io::Error::other(format!("Stream: {s}. Error {e}")))
                         }
                     }
                 };

crates/sigma-stream/src/tests.rs

@@ -0,0 +1,85 @@
+use super::*;
+use bytes::BytesMut;
+
+#[test]
+fn test_sse_decoder_valid_data() {
+    let mut decoder = SseDecoder::<serde_json::Value>::new();
+    let mut buf = BytesMut::new();
+
+    // Add a valid SSE message
+    buf.extend_from_slice(b"data: {\"test\": \"value\"}\n\n");
+
+    let result = decoder.decode(&mut buf).unwrap();
+    assert!(result.is_some());
+
+    let value = result.unwrap();
+    assert_eq!(value["test"], "value");
+
+    // Buffer should be empty after decoding
+    assert!(buf.is_empty());
+}
+
+#[test]
+fn test_sse_decoder_keep_alive() {
+    let mut decoder = SseDecoder::<serde_json::Value>::new();
+    let mut buf = BytesMut::new();
+
+    // Add a keep-alive message
+    buf.extend_from_slice(b"data: :\n\n");
+
+    let result = decoder.decode(&mut buf).unwrap();
+    assert!(result.is_none()); // Keep-alive should return None
+
+    // Buffer should be empty after decoding
+    assert!(buf.is_empty());
+}
+
+#[test]
+fn test_sse_decoder_partial_message() {
+    let mut decoder = SseDecoder::<serde_json::Value>::new();
+    let mut buf = BytesMut::new();
+
+    // Add partial message (no double newline)
+    buf.extend_from_slice(b"data: {\"test\": \"value\"}\n");
+
+    let result = decoder.decode(&mut buf).unwrap();
+    assert!(result.is_none()); // Should need more data
+
+    // Buffer should still contain the partial message
+    assert!(!buf.is_empty());
+
+    // Complete the message
+    buf.extend_from_slice(b"\n");
+
+    let result = decoder.decode(&mut buf).unwrap();
+    assert!(result.is_some());
+}
+
+#[test]
+fn test_sse_decoder_invalid_json() {
+    let mut decoder = SseDecoder::<serde_json::Value>::new();
+    let mut buf = BytesMut::new();
+
+    // Add invalid JSON
+    buf.extend_from_slice(b"data: {invalid json}\n\n");
+
+    let result = decoder.decode(&mut buf);
+    assert!(result.is_err());
+}
+
+#[test]
+fn test_sse_decoder_invalid_utf8() {
+    let mut decoder = SseDecoder::<serde_json::Value>::new();
+    let mut buf = BytesMut::new();
+
+    // Add invalid UTF-8
+    buf.extend_from_slice(b"data: ");
+    buf.extend_from_slice(&[0xFF, 0xFE]); // Invalid UTF-8
+    buf.extend_from_slice(b"\n\n");
+
+    let result = decoder.decode(&mut buf).unwrap();
+    assert!(result.is_none()); // Should skip bad data
+
+    // Buffer should be empty after skipping
+    assert!(buf.is_empty());
+}