diff --git a/.github/ci.yml b/.github/workflows/ci.yml similarity index 54% rename from .github/ci.yml rename to .github/workflows/ci.yml index a3ca06a..90ea881 100644 --- a/.github/ci.yml +++ b/.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 diff --git a/Cargo.lock b/Cargo.lock index 273148a..021da38 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3211,6 +3211,7 @@ dependencies = [ "serde", "serde_json", "sigma-types", + "tokio", "tokio-util", ] @@ -3222,6 +3223,15 @@ dependencies = [ "serde", ] +[[package]] +name = "signal-hook-registry" +version = "1.4.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9203b8055f63a2a00e2f593bb0510367fe707d7ff1e5c872de2f537b339e5410" +dependencies = [ + "libc", +] + [[package]] name = "signature" version = "2.2.0" @@ -3532,7 +3542,9 @@ dependencies = [ "io-uring", "libc", "mio", + "parking_lot", "pin-project-lite", + "signal-hook-registry", "slab", "socket2", "tokio-macros", diff --git a/Cargo.toml b/Cargo.toml index 23ca30b..8a1b5ee 100644 --- a/Cargo.toml +++ b/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" \ No newline at end of file +sigma-types = { git = "ssh://git@github.com/essential-contributions/sigma-node.git"} +tokio = { version = "1.44.0", features = ["full"] } +tokio-util = "0.7.15" diff --git a/config/test.yaml b/config/test.yaml new file mode 100644 index 0000000..b675b3f --- /dev/null +++ b/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 diff --git a/crates/sigma-stream/Cargo.toml b/crates/sigma-stream/Cargo.toml index b533eea..ecee1ab 100644 --- a/crates/sigma-stream/Cargo.toml +++ b/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 \ No newline at end of file +tokio.workspace = true +tokio-util.workspace = true diff --git a/crates/sigma-stream/src/lib.rs b/crates/sigma-stream/src/lib.rs index 29a3d20..f7bbff8 100644 --- a/crates/sigma-stream/src/lib.rs +++ b/crates/sigma-stream/src/lib.rs @@ -1,9 +1,8 @@ #![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}, @@ -11,32 +10,33 @@ use tokio_util::{ 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; /// 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>, std::io::Error> { +) -> Result, 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::>::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::(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}"))) } } }; diff --git a/crates/sigma-stream/src/tests.rs b/crates/sigma-stream/src/tests.rs new file mode 100644 index 0000000..6e7b9a6 --- /dev/null +++ b/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::::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::::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::::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::::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::::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()); +} diff --git a/crates/sigma-stream/tests/common/mod.rs b/crates/sigma-stream/tests/common/mod.rs new file mode 100644 index 0000000..704b2d5 --- /dev/null +++ b/crates/sigma-stream/tests/common/mod.rs @@ -0,0 +1,91 @@ +use std::net::TcpStream; +use std::process::{Child, Command, Stdio}; +use std::thread; +use std::time::Duration; + +pub struct SigmaNode { + process: Child, +} + +impl SigmaNode { + /// Start sigma-node if it's available in PATH + pub fn start() -> Result { + // Check if sigma-node is available + if !Self::is_available() { + return Err("sigma-node not found in PATH. Please ensure it's installed.".to_string()); + } + + // Start sigma-node + let mut cmd = Command::new("sigma-node"); + cmd.arg("--bind-address").arg("127.0.0.1:3000"); + + // Add config if it exists + if std::path::Path::new("config/test.yaml").exists() { + cmd.arg("--config").arg("config/test.yaml"); + } + + cmd.stdout(Stdio::null()).stderr(Stdio::null()); + + let process = cmd + .spawn() + .map_err(|e| format!("Failed to start sigma-node: {e}"))?; + + // Wait for sigma-node to be ready + Self::wait_for_ready()?; + + Ok(SigmaNode { process }) + } + + /// Check if sigma-node is available in PATH + pub fn is_available() -> bool { + Command::new("sigma-node").arg("--version").output().is_ok() + } + + /// Wait for sigma-node to be ready + fn wait_for_ready() -> Result<(), String> { + println!("Waiting for sigma-node to start..."); + + for i in 0..30 { + if TcpStream::connect("127.0.0.1:3000").is_ok() { + println!("Sigma-node is ready!"); + return Ok(()); + } + thread::sleep(Duration::from_secs(1)); + + if i % 5 == 0 { + println!("Still waiting for sigma-node... ({i}s)"); + } + } + + Err("Timeout waiting for sigma-node to start".to_string()) + } +} + +impl Drop for SigmaNode { + fn drop(&mut self) { + println!("Stopping sigma-node..."); + + // Try graceful shutdown first + if let Err(e) = self.process.kill() { + eprintln!("Error killing sigma-node: {e}"); + } + + // Wait for process to exit + let _ = self.process.wait(); + } +} + +/// Setup function for tests that need sigma-node +pub fn setup_sigma_node() -> Option { + match SigmaNode::start() { + Ok(node) => { + println!("Successfully started sigma-node for tests"); + Some(node) + } + Err(e) => { + println!("WARNING: {e}"); + println!("Skipping tests that require sigma-node"); + None + } + } +} diff --git a/crates/sigma-stream/tests/stream_events.rs b/crates/sigma-stream/tests/stream_events.rs new file mode 100644 index 0000000..7e1ae14 --- /dev/null +++ b/crates/sigma-stream/tests/stream_events.rs @@ -0,0 +1,83 @@ +mod common; + +use futures::StreamExt; +use reqwest::Url; +use sigma_stream::stream_events; +use tokio::time::{Duration, timeout}; + +#[tokio::test] +async fn test_stream_events_from_real_endpoint() { + // Try to start sigma-node + let _node = match common::setup_sigma_node() { + Some(node) => node, + None => { + println!("Test skipped: sigma-node not available"); + return; + } + }; + + // Now run the actual test - sigma-node is guaranteed to be running + let node_url = Url::parse("http://localhost:3000").expect("Invalid URL"); + let start_height = 0; + + println!("Connecting to Sigma node at: {node_url}"); + println!("Starting from event height: {start_height}"); + + let stream_result = stream_events(&node_url, start_height).await; + + let mut stream = match stream_result { + Ok(s) => { + println!("Successfully connected to event stream"); + s + } + Err(e) => { + panic!("Failed to create stream: {e}"); + } + }; + + // Set a timeout for the test + let timeout_duration = Duration::from_secs(30); + + // Try to receive at least one event + let result = timeout(timeout_duration, async { + let mut events_received = 0; + while let Some(event_result) = stream.next().await { + match event_result { + Ok(signed_event) => { + assert!( + signed_event.data.height >= start_height as u64, + "Event height should be >= start height" + ); + println!( + "Received signed event at height {}: stream_type={:?}, data_len={}, signature_len={}", + signed_event.data.height, + signed_event.data.event_data.stream_type, + signed_event.data.event_data.data.len(), + signed_event.signature.len() + ); + assert!(!signed_event.signature.is_empty(), "Signature should not be empty"); + events_received += 1; + if events_received >= 3 { + break; + } + } + Err(e) => { + panic!("Error receiving event: {e}"); + } + } + } + assert!(events_received > 0, "Should have received at least one event"); + events_received + }).await; + + match result { + Ok(count) => { + println!("Successfully received {count} events"); + } + Err(_) => { + panic!("Test timed out waiting for events"); + } + } + + // Node will be automatically stopped when _node goes out of scope +} diff --git a/flake.lock b/flake.lock new file mode 100644 index 0000000..b2985d5 --- /dev/null +++ b/flake.lock @@ -0,0 +1,80 @@ +{ + "nodes": { + "nixpkgs": { + "locked": { + "lastModified": 1752950548, + "narHash": "sha256-NS6BLD0lxOrnCiEOcvQCDVPXafX1/ek1dfJHX1nUIzc=", + "owner": "nixos", + "repo": "nixpkgs", + "rev": "c87b95e25065c028d31a94f06a62927d18763fdf", + "type": "github" + }, + "original": { + "owner": "nixos", + "ref": "nixos-unstable", + "repo": "nixpkgs", + "type": "github" + } + }, + "root": { + "inputs": { + "nixpkgs": "nixpkgs", + "sigma-node": "sigma-node", + "systems": "systems_2" + } + }, + "sigma-node": { + "inputs": { + "nixpkgs": [ + "nixpkgs" + ], + "systems": "systems" + }, + "locked": { + "lastModified": 1753051747, + "narHash": "sha256-LuBNQljNzTzI/DFJu17NXd8sfGithV7iGWMIPZO5IWM=", + "ref": "refs/heads/main", + "rev": "dec3a6f3a97d9228d3ec5791abc1d3a4d6ec5b2d", + "revCount": 15, + "type": "git", + "url": "ssh://git@github.com/essential-contributions/sigma-node" + }, + "original": { + "type": "git", + "url": "ssh://git@github.com/essential-contributions/sigma-node" + } + }, + "systems": { + "locked": { + "lastModified": 1681028828, + "narHash": "sha256-Vy1rq5AaRuLzOxct8nz4T6wlgyUR7zLU309k9mBC768=", + "owner": "nix-systems", + "repo": "default", + "rev": "da67096a3b9bf56a91d16901293e51ba5b49a27e", + "type": "github" + }, + "original": { + "owner": "nix-systems", + "repo": "default", + "type": "github" + } + }, + "systems_2": { + "locked": { + "lastModified": 1681028828, + "narHash": "sha256-Vy1rq5AaRuLzOxct8nz4T6wlgyUR7zLU309k9mBC768=", + "owner": "nix-systems", + "repo": "default", + "rev": "da67096a3b9bf56a91d16901293e51ba5b49a27e", + "type": "github" + }, + "original": { + "owner": "nix-systems", + "repo": "default", + "type": "github" + } + } + }, + "root": "root", + "version": 7 +} diff --git a/flake.nix b/flake.nix new file mode 100644 index 0000000..7f64b88 --- /dev/null +++ b/flake.nix @@ -0,0 +1,33 @@ +{ + description = "A nix flake for your Rust project with sigma-node integration"; + + inputs = { + nixpkgs.url = "github:nixos/nixpkgs/nixos-unstable"; + systems.url = "github:nix-systems/default"; + # Use SSH URL for private repo access + sigma-node = { + url = "git+ssh://git@github.com/essential-contributions/sigma-node"; + inputs.nixpkgs.follows = "nixpkgs"; + }; + }; + + outputs = { self, nixpkgs, systems, sigma-node }: + let + # Helper to generate outputs for all systems + perSystemPkgs = f: + nixpkgs.lib.genAttrs (import systems) ( + system: + let + pkgs = import nixpkgs { inherit system; }; + in + f pkgs + ); + in + { + devShells = perSystemPkgs (pkgs: { + default = pkgs.callPackage ./shell.nix { + inherit (sigma-node.packages.${pkgs.system}) sigma-node; + }; + }); + }; +} diff --git a/shell.nix b/shell.nix new file mode 100644 index 0000000..ca1fce0 --- /dev/null +++ b/shell.nix @@ -0,0 +1,48 @@ +# Development shell for sigma-tools +{ + mkShell, + cargo, + rustc, + clippy, + rustfmt, + rust-analyzer, + cargo-toml-lint, + openssl, + pkg-config, + netcat, + sigma-node, +}: +mkShell { + inputsFrom = [ + # Include dependencies from our packages if needed + ]; + + buildInputs = [ + # Rust toolchain + cargo + rustc + clippy + rustfmt + rust-analyzer + cargo-toml-lint + + # Dependencies + openssl + openssl.dev + pkg-config + + # Tools + sigma-node + netcat + ]; + + # Environment variables + OPENSSL_NO_VENDOR = 1; + + shellHook = '' + echo "Sigma-tools development shell" + echo "Available commands:" + echo " sigma-node - Run sigma-node" + echo " cargo test - Run tests (will auto-start sigma-node)" + ''; +}