Repository navigation
Conversation
Remove obsolete socket transport fields from `TestIngestdConfig` and `TestIngestdHandle` so the shared fixture API reflects the actual NATS-only startup path.
Replace the socket/gRPC-oriented architecture summary with the direct JetStream flow and confirmation model now implemented in code.
Update the shared processor CLI help so `NATS_URL` is the primary transport signal and the socket path is clearly transitional.
Delete the remaining socket-path config and CLI wiring and make the NATS URL the only transport setting in the processor entry points. Co-Authored-By: Claude <noreply@anthropic.com>
… SDK Drop the dead Redis URL field and the matching test configuration now that the runtime is NATS-only. Co-Authored-By: Claude <noreply@anthropic.com>
Replace the fixed startup delay with a real stream-availability check before tests assume ingestd and JetStream are ready. Co-Authored-By: Claude <noreply@anthropic.com>
- add `AutomatonEventHandler` as a confirmed-event bridge - initialize `LeaseManager` and `ConfirmationBuffer` during `StreamProcessorRunner` startup for automata
…ases Add `jetstream_e2e_integration_test.rs` to prove the new startup path while restoring the analytics/search automaton aliases expected by the workspace.
Refresh the phase/status language in `docs/way.md` so completed migration scope and follow-up runtime work are separated cleanly.
Sinity
added
type:docs
|
You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard. |
|
Note
|
| Cohort / File(s) | Summary |
|---|---|
Workspace & Documentation Cargo.toml, README.md, docs/way.md |
Cargo comments updated to note type aliases and refactoring needs; README architecture updated to describe NATS JetStream as primary ingestion bus with Satellites publishing to events.raw.* and ingestd as consumer/archiver; implementation status documented with phase completion and test coverage added. |
Configuration & CLI Interfaces crate/lib/sinex-satellite-sdk/src/cli.rs, crate/lib/sinex-satellite-sdk/src/config.rs, crate/lib/sinex-satellite-sdk/src/lib.rs |
Removed ingest_socket_path and redis_url fields; replaced with required nats_url field (defaulting to nats://localhost:4222). Simplified conditional transport selection and validation logic to always use NATS. |
Event Handling & Publisher crate/lib/sinex-satellite-sdk/src/automaton_event_handler.rs, crate/lib/sinex-satellite-sdk/src/nats_publisher.rs |
Added new AutomatonEventHandler struct with thread-safe state tracking for confirmed event IDs and processing counters, supporting test introspection. Exposed NATS client reference via nats_client() accessor on NatsPublisher. |
Stream Processing & Lease Management crate/lib/sinex-satellite-sdk/src/stream_processor.rs |
Extended StreamProcessorContext and StreamProcessorRunner with optional lease_manager and confirmation_buffer fields. For automata, conditionally creates LeaseManager and switches to LeaderStandby processing model; checks leadership before executing continuous scan and stops lease manager on shutdown. |
Integration Tests crate/core/sinex-ingestd/tests/jetstream_e2e_integration_test.rs |
Added two comprehensive async tests: one verifying end-to-end event flow from publication through database persistence and automaton confirmation handling; another validating idempotency by publishing identical events and asserting single database record. |
Test Infrastructure & Utilities crate/lib/sinex-satellite-sdk/tests/integration/*, crate/lib/sinex-test-utils/src/satellite_management_utils.rs |
Updated test configuration builders to use nats_url instead of socket paths. Changed ingestd readiness detection from socket file existence to NATS stream availability. Removed socket-path fields from TestIngestdConfig and TestIngestdHandle. |
Processor Type Aliases crate/satellites/sinex-analytics-automaton/src/lib.rs, crate/satellites/sinex-search-automaton/src/lib.rs |
Added AnalyticsProcessor and SearchProcessor type aliases mapping to AnalyticsAutomaton and SearchAutomaton respectively for macro compatibility. |
Sequence Diagram(s)
sequenceDiagram
participant Publisher as NatsPublisher
participant JStream as NATS JetStream<br/>(events.raw.*)
participant Ingestd as ingestd<br/>JetStreamConsumer
participant DB as Postgres<br/>(core.events)
participant ConfirmStream as NATS JetStream<br/>(events.confirmations.*)
participant Automaton as Automaton<br/>JetStreamEventConsumer
participant Handler as AutomatonEventHandler
Publisher->>JStream: Publish event with<br/>NATS-Msg-Id (idempotency)
Ingestd->>JStream: Consume event
Ingestd->>DB: Validate & persist event
Ingestd->>ConfirmStream: Publish confirmation
Automaton->>ConfirmStream: Consume confirmation
Automaton->>Handler: Handle confirmed event
Handler->>Handler: Record event ID &<br/>increment counter
Estimated code review effort
🎯 4 (Complex) | ⏱️ ~60 minutes
Poem
🐰 A message hops through JetStream's flow,
From raw events to Postgres below,
Automata confirm with idempotent grace,
NATS choreography sets the new pace! 🎀✨
🚥 Pre-merge checks | ✅ 1
✅ Passed checks (1 passed)
| Check name | Status | Explanation |
|---|---|---|
| Title check | ✅ Passed | The title follows Conventional Commits format with type(scope): subject, correctly using docs(testing) as the scoped type and providing a clear, specific description of the changes related to NATS-only ingest and JetStream e2e testing. |
✏️ Tip: You can configure your own custom pre-merge checks in the settings.
✨ Finishing Touches
📝 Generate docstrings
- Create stacked PR
- Commit on current branch
🧪 Generate unit tests (beta)
- Create PR with unit tests
- Commit unit tests in branch
feature/docs/nats-only-ingest-testing
Comment @coderabbitai help to get the list of available commands and usage tips.
Sinity
added a commit
that referenced
this pull request
Apr 9, 2026
Finish the socket-era cleanup in the public docs and shared test helpers, then
add the runtime bridge needed for a real JetStream end-to-end test.
test-utils:
- remove `socket_path` from `crate/lib/sinex-test-utils/src/satellite_management_utils.rs` and wait for NATS readiness instead of sleeping, so ingestd test fixtures describe the transport that actually exists
- keep `TestIngestdHandle` focused on stream metadata and in-process service control rather than carrying obsolete socket-era fields
sinex-satellite-sdk:
- deprecate and then remove `ingest_socket_path` from `src/{cli.rs,config.rs,lib.rs}`, replacing it with a required `nats_url` path and deleting the unused Redis config fields
- add `crate/lib/sinex-satellite-sdk/src/automaton_event_handler.rs` and wire `LeaseManager` plus `ConfirmationBuffer` into `StreamProcessorRunner`, giving automata a confirmed-event bridge during startup
- restore the analytics and search automaton aliases and add JetStream end-to-end coverage so the new startup path is exercised from the test suite
docs:
- rewrite `README.md` around the NATS-only ingest flow and update `docs/way.md` so the migration phases and remaining JetStream work match the code
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 10
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
crate/lib/sinex-satellite-sdk/src/config.rs (1)
220-226:⚠️ Potential issue | 🟡 MinorDocumentation references removed environment variables.
The docstring mentions
SINEX_INGEST_SOCKETandSINEX_REDIS_URLwhich no longer exist. Update to reflect the actualNATS_URLvariable.📝 Proposed doc fix
/// # Environment Variables /// - `SINEX_LOG_LEVEL`: Log level (default: "info") - /// - `SINEX_INGEST_SOCKET`: Socket path (default: "/run/sinex/ingest.sock") - /// - `SINEX_REDIS_URL`: Redis URL (default: "redis://localhost:6379") + /// - `NATS_URL`: NATS URL (default: "nats://localhost:4222") /// - `DATABASE_URL`: PostgreSQL URL (optional) /// - `SINEX_DB_POOL_SIZE`: Pool size (default: 10)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@crate/lib/sinex-satellite-sdk/src/config.rs` around lines 220 - 226, The doc comment in config.rs incorrectly lists removed env vars `SINEX_INGEST_SOCKET` and `SINEX_REDIS_URL`; update the docstring associated with the Config (or the top-level config module) to remove those two entries and instead document the actual `NATS_URL` environment variable (include a short description and default if applicable, e.g., NATS server URL, default "nats://localhost:4222"). Ensure the surrounding list still includes `SINEX_LOG_LEVEL`, `DATABASE_URL`, `SINEX_DB_POOL_SIZE`, `SINEX_WORK_DIR`, and `SINEX_DRY_RUN` and that the doc comment stays syntactically valid rustdoc for the Config-related symbol (Config struct or module) so tooling picks up the change.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@Cargo.toml`:
- Around line 28-30: Update the inline TODO comment "Type aliases added but
compilation errors remain" to reference a concrete tracking item (issue number
or ADR URL) and a brief re-enablement criterion; find the comment text in
Cargo.toml (the lines containing "Type aliases added but compilation errors
remain" and the commented crate entries) and replace it with a TODO that
includes the issue/ADR identifier (e.g., GH-1234 or a full ADR link) plus a
short note on what must be resolved before uncommenting the crates (e.g.,
specific compilation errors or tests to pass).
In `@crate/core/sinex-ingestd/tests/jetstream_e2e_integration_test.rs`:
- Around line 188-191: The current cleanup just drops the JoinHandles
(ingestd_handle and automaton_handle) which doesn't stop the spawned tasks; call
abort() on both JoinHandle values (ingestd_handle.abort() and
automaton_handle.abort()) and then await their completions (await the handles)
to ensure the tasks are cancelled and any panics are observed before the test
exits; update the cleanup section that currently drops ingestd_handle and
automaton_handle to explicitly abort and await them instead.
- Around line 221-222: The spawned task handle _ingestd_handle created from
tokio::spawn(async move { ingestd_consumer.run().await }) is never cleaned up at
the end of the test; ensure the background consumer is stopped by either
awaiting the handle to finish or aborting it when the test completes (e.g., call
_ingestd_handle.abort() or await _ingestd_handle with a timeout and then abort
if still running) so ingestd_consumer.run() is not left running after the test.
In `@crate/lib/sinex-satellite-sdk/src/automaton_event_handler.rs`:
- Around line 68-72: The processed_event_ids Vec grows unbounded; change its
type to a bounded buffer (e.g., VecDeque<Ulid>) or enforce a capacity when
writing to it to avoid unbounded memory growth. Update the field
processed_event_ids to Arc<RwLock<VecDeque<Ulid>>> (import
std::collections::VecDeque) and in the handler where you currently push
provisional.event_id, replace the push with: if ids.len() >= MAX_TRACKED_IDS {
ids.pop_front(); } ids.push_back(provisional.event_id); or alternatively
truncate the Vec to the last N entries after pushing; define a MAX_TRACKED_IDS
const and apply this logic where ids are mutated.
In `@crate/lib/sinex-satellite-sdk/src/cli.rs`:
- Around line 507-520: Duplicate NATS setup in the Service and Scan handlers
should be extracted into a shared async helper (e.g., create_nats_transport)
that accepts a &str nats_url, logs the URL (info!), calls async_nats::connect,
wraps errors with .context("Failed to connect to NATS") returning
color_eyre::eyre::Result<EventTransport>, constructs a
crate::nats_publisher::NatsPublisher via NatsPublisher::new and returns
crate::event_processor::EventTransport::Nats(std::sync::Arc::new(nats_publisher));
replace the duplicated blocks in both handlers with calls to this helper.
In `@crate/lib/sinex-satellite-sdk/src/config.rs`:
- Around line 328-330: SatelliteFigmentConfig and SatelliteConfig diverge:
SatelliteFigmentConfig defines socket_path and redis_url (default
/tmp/sinex-ingestd.sock and redis://localhost:6379) while SatelliteConfig uses
nats_url; align them by updating SatelliteFigmentConfig in figment_config.rs to
include nats_url with the same default as default_nats_url() (and remove or mark
redis_url deprecated if unused), keep socket_path in place because
sensd_client.rs depends on it, and remove any unused redis_url references to
avoid confusion; ensure the struct name SatelliteFigmentConfig and the field
names socket_path, redis_url, and nats_url are updated consistently and
tests/usage updated accordingly.
In `@crate/lib/sinex-satellite-sdk/src/stream_processor.rs`:
- Around line 1006-1016: Replace the fixed 2s sleep after lease_manager.start()
with a readiness loop that polls lease_manager.status().await until it indicates
leadership (or a ready state) or a deadline/timeout is reached; continue to call
lease_manager.start().await then loop calling lease_manager.status().await with
a small exponential/backoff delay (e.g., 100-500ms) and a configurable timeout,
logging interim statuses and failing/returning an error if the timeout elapses,
so that the automaton only proceeds when self.lease_manager reports the initial
lease acquisition.
- Around line 1026-1036: The function run_automaton_continuous_mode currently
checks lease_manager.status() once and returns early for
ProcessingModel::LeaderStandby, causing the standby to exit instead of waiting;
change the logic in run_automaton_continuous_mode so that when
self.processing_model == crate::ProcessingModel::LeaderStandby and a
lease_manager exists you do not return immediately on non-Leader status but
instead wait for leadership (either by subscribing to a lease status change
event on lease_manager, calling a provided wait_for_leadership-like API, or
implementing a short async loop/poll that rechecks lease_manager.status().await
until it equals crate::LeaseStatus::Leader), only proceeding to start processing
after confirming LeaseStatus::Leader and keeping the function alive to re-check
on loss of leadership (i.e., stop processing and return to waiting if status
changes away from Leader).
In `@docs/way.md`:
- Around line 168-172: The numbered list around the "Migratory template for each
satellite" block is missing surrounding blank lines and causing markdown lint
errors; edit the section containing the list (mentions of AcquisitionManager,
Stage-as-You-Go, and NatsPublisher) to ensure there's a blank line before and
after the numbered list and any nearby headings or fenced code blocks, and
add/adjust blank lines between list items and adjacent headings/code fences so
the markdown linter no longer reports "missing blank lines around headings and
fenced code blocks."
In `@README.md`:
- Line 103: Add a blank line immediately before the markdown heading "### ✅
Implemented (Working Code)" to satisfy MD022 (ensure there is a separating empty
line above the heading).
---
Outside diff comments:
In `@crate/lib/sinex-satellite-sdk/src/config.rs`:
- Around line 220-226: The doc comment in config.rs incorrectly lists removed
env vars `SINEX_INGEST_SOCKET` and `SINEX_REDIS_URL`; update the docstring
associated with the Config (or the top-level config module) to remove those two
entries and instead document the actual `NATS_URL` environment variable (include
a short description and default if applicable, e.g., NATS server URL, default
"nats://localhost:4222"). Ensure the surrounding list still includes
`SINEX_LOG_LEVEL`, `DATABASE_URL`, `SINEX_DB_POOL_SIZE`, `SINEX_WORK_DIR`, and
`SINEX_DRY_RUN` and that the doc comment stays syntactically valid rustdoc for
the Config-related symbol (Config struct or module) so tooling picks up the
change.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: 321ebbee-31a8-4042-90c8-cc9c5e7371b7
📒 Files selected for processing (15)
Cargo.tomlREADME.mdcrate/core/sinex-ingestd/tests/jetstream_e2e_integration_test.rscrate/lib/sinex-satellite-sdk/src/automaton_event_handler.rscrate/lib/sinex-satellite-sdk/src/cli.rscrate/lib/sinex-satellite-sdk/src/config.rscrate/lib/sinex-satellite-sdk/src/lib.rscrate/lib/sinex-satellite-sdk/src/nats_publisher.rscrate/lib/sinex-satellite-sdk/src/stream_processor.rscrate/lib/sinex-satellite-sdk/tests/integration/satellite_architecture_test.rscrate/lib/sinex-satellite-sdk/tests/integration/satellite_lifecycle_test.rscrate/lib/sinex-test-utils/src/satellite_management_utils.rscrate/satellites/sinex-analytics-automaton/src/lib.rscrate/satellites/sinex-search-automaton/src/lib.rsdocs/way.md
Comment on lines
+28
to
+30
| # Note: Type aliases added but compilation errors remain - needs refactoring | ||
| # "crate/satellites/sinex-analytics-automaton", | ||
| # "crate/satellites/sinex-search-automaton", |
There was a problem hiding this comment.
🧹 Nitpick | 🔵 Trivial
Consider linking this TODO comment to a concrete tracking issue.
The “compilation errors remain” note is useful, but adding an issue/ADR reference here would prevent this from becoming stale and make re-enablement criteria explicit.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@Cargo.toml` around lines 28 - 30, Update the inline TODO comment "Type
aliases added but compilation errors remain" to reference a concrete tracking
item (issue number or ADR URL) and a brief re-enablement criterion; find the
comment text in Cargo.toml (the lines containing "Type aliases added but
compilation errors remain" and the commented crate entries) and replace it with
a TODO that includes the issue/ADR identifier (e.g., GH-1234 or a full ADR link)
plus a short note on what must be resolved before uncommenting the crates (e.g.,
specific compilation errors or tests to pass).
Comment on lines
+188
to
+191
| // Cleanup | ||
| drop(ingestd_handle); | ||
| drop(automaton_handle); | ||
|
|
There was a problem hiding this comment.
Dropping task handles does not cancel spawned tasks.
drop(ingestd_handle) and drop(automaton_handle) simply release the JoinHandle without aborting the underlying tasks. The tasks will continue running until the test process exits.
🧹 Proposed cleanup
// Cleanup
- drop(ingestd_handle);
- drop(automaton_handle);
+ ingestd_handle.abort();
+ automaton_handle.abort();
+ // Optionally wait for cleanup
+ let _ = ingestd_handle.await;
+ let _ = automaton_handle.await;📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| // Cleanup | |
| drop(ingestd_handle); | |
| drop(automaton_handle); | |
| // Cleanup | |
| ingestd_handle.abort(); | |
| automaton_handle.abort(); | |
| // Optionally wait for cleanup | |
| let _ = ingestd_handle.await; | |
| let _ = automaton_handle.await; |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@crate/core/sinex-ingestd/tests/jetstream_e2e_integration_test.rs` around
lines 188 - 191, The current cleanup just drops the JoinHandles (ingestd_handle
and automaton_handle) which doesn't stop the spawned tasks; call abort() on both
JoinHandle values (ingestd_handle.abort() and automaton_handle.abort()) and then
await their completions (await the handles) to ensure the tasks are cancelled
and any panics are observed before the test exits; update the cleanup section
that currently drops ingestd_handle and automaton_handle to explicitly abort and
await them instead.
Comment on lines
+221
to
+222
| let _ingestd_handle = tokio::spawn(async move { ingestd_consumer.run().await }); | ||
| tokio::time::sleep(Duration::from_secs(1)).await; |
There was a problem hiding this comment.
Same task cleanup issue in idempotency test.
The _ingestd_handle is not aborted on completion.
🧹 Add cleanup at end of test
info!("🎉 Idempotency test PASSED - 3 publishes → 1 database row");
+ _ingestd_handle.abort();
+
Ok(())
}🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@crate/core/sinex-ingestd/tests/jetstream_e2e_integration_test.rs` around
lines 221 - 222, The spawned task handle _ingestd_handle created from
tokio::spawn(async move { ingestd_consumer.run().await }) is never cleaned up at
the end of the test; ensure the background consumer is stopped by either
awaiting the handle to finish or aborting it when the test completes (e.g., call
_ingestd_handle.abort() or await _ingestd_handle with a timeout and then abort
if still running) so ingestd_consumer.run() is not left running after the test.
Comment on lines
+68
to
+72
| // Track the event ID | ||
| { | ||
| let mut ids = self.processed_event_ids.write().await; | ||
| ids.push(provisional.event_id); | ||
| } |
There was a problem hiding this comment.
🧹 Nitpick | 🔵 Trivial
processed_event_ids vector grows unbounded.
For long-running automata, the Vec<Ulid> will grow indefinitely. Consider adding a capacity limit or periodic cleanup for production use.
♻️ Optional: Add capacity management
// Option 1: Use a bounded ring buffer
use std::collections::VecDeque;
processed_event_ids: Arc<RwLock<VecDeque<Ulid>>>,
// In handle_confirmed:
let mut ids = self.processed_event_ids.write().await;
if ids.len() >= MAX_TRACKED_IDS {
ids.pop_front();
}
ids.push_back(provisional.event_id);Since this is documented as a transitional adapter primarily for testing, this may be acceptable as-is for now.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@crate/lib/sinex-satellite-sdk/src/automaton_event_handler.rs` around lines 68
- 72, The processed_event_ids Vec grows unbounded; change its type to a bounded
buffer (e.g., VecDeque<Ulid>) or enforce a capacity when writing to it to avoid
unbounded memory growth. Update the field processed_event_ids to
Arc<RwLock<VecDeque<Ulid>>> (import std::collections::VecDeque) and in the
handler where you currently push provisional.event_id, replace the push with: if
ids.len() >= MAX_TRACKED_IDS { ids.pop_front(); }
ids.push_back(provisional.event_id); or alternatively truncate the Vec to the
last N entries after pushing; define a MAX_TRACKED_IDS const and apply this
logic where ids are mutated.
Comment on lines
+507
to
+520
| // Create NATS transport for event publishing | ||
| info!(nats_url = %args.nats_url, "Using NATS for event publishing"); | ||
|
|
||
| // Connect to NATS | ||
| let nats_client = async_nats::connect(&nats_url) | ||
| .await | ||
| .context("Failed to connect to NATS")?; | ||
| // Connect to NATS | ||
| let nats_client = async_nats::connect(&args.nats_url) | ||
| .await | ||
| .context("Failed to connect to NATS")?; | ||
|
|
||
| // Create NATS publisher | ||
| let nats_publisher = crate::nats_publisher::NatsPublisher::new(nats_client); | ||
| // Create NATS publisher | ||
| let nats_publisher = crate::nats_publisher::NatsPublisher::new(nats_client); | ||
|
|
||
| crate::event_processor::EventTransport::Nats(std::sync::Arc::new( | ||
| nats_publisher, | ||
| )) | ||
| } else { | ||
| return Err(eyre::eyre!( | ||
| "NATS URL is required. Use --nats-url or set NATS_URL environment variable. gRPC is no longer supported." | ||
| )); | ||
| }; | ||
| let transport = crate::event_processor::EventTransport::Nats(std::sync::Arc::new( | ||
| nats_publisher, | ||
| )); |
There was a problem hiding this comment.
🧹 Nitpick | 🔵 Trivial
Consider extracting duplicated NATS transport setup.
The NATS connection and publisher creation logic is nearly identical in both Service and Scan command handlers.
♻️ Optional: Extract helper function
async fn create_nats_transport(nats_url: &str) -> color_eyre::eyre::Result<EventTransport> {
info!(nats_url = %nats_url, "Using NATS for event publishing");
let nats_client = async_nats::connect(nats_url)
.await
.context("Failed to connect to NATS")?;
let nats_publisher = crate::nats_publisher::NatsPublisher::new(nats_client);
Ok(crate::event_processor::EventTransport::Nats(std::sync::Arc::new(nats_publisher)))
}This could be used in both handlers to reduce duplication.
Also applies to: 630-643
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@crate/lib/sinex-satellite-sdk/src/cli.rs` around lines 507 - 520, Duplicate
NATS setup in the Service and Scan handlers should be extracted into a shared
async helper (e.g., create_nats_transport) that accepts a &str nats_url, logs
the URL (info!), calls async_nats::connect, wraps errors with .context("Failed
to connect to NATS") returning color_eyre::eyre::Result<EventTransport>,
constructs a crate::nats_publisher::NatsPublisher via NatsPublisher::new and
returns
crate::event_processor::EventTransport::Nats(std::sync::Arc::new(nats_publisher));
replace the duplicated blocks in both handlers with calls to this helper.
Comment on lines
+328
to
330
| fn default_nats_url() -> String { | ||
| "nats://localhost:4222".to_string() | ||
| } |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Check if SatelliteFigmentConfig is still used
rg -n "SatelliteFigmentConfig" --type rustRepository: Sinity/sinex
Length of output: 1033
🏁 Script executed:
cat -n crate/lib/sinex-satellite-sdk/src/figment_config.rs | sed -n '1,100p'Repository: Sinity/sinex
Length of output: 4177
🏁 Script executed:
cat -n crate/lib/sinex-satellite-sdk/src/config.rs | head -100Repository: Sinity/sinex
Length of output: 4385
🏁 Script executed:
rg -n "fn default_socket_path|fn default_redis_url" crate/lib/sinex-satellite-sdk/src/figment_config.rs -A 2Repository: Sinity/sinex
Length of output: 238
🏁 Script executed:
rg -n "fn default_log_level|fn default_nats_url|fn default_work_dir" crate/lib/sinex-satellite-sdk/src/ -A 2Repository: Sinity/sinex
Length of output: 1288
🏁 Script executed:
head -50 crate/lib/sinex-satellite-sdk/src/config.rs | grep -E "^//|legacy|old|deprecated"Repository: Sinity/sinex
Length of output: 1327
🏁 Script executed:
head -50 crate/lib/sinex-satellite-sdk/src/figment_config.rs | grep -E "^//|legacy|old|deprecated"Repository: Sinity/sinex
Length of output: 291
🏁 Script executed:
find crate/lib/sinex-satellite-sdk -name "*.rs" -type f | xargs rg -l "SatelliteConfig|SatelliteFigmentConfig" | head -5Repository: Sinity/sinex
Length of output: 422
🏁 Script executed:
rg -n "socket_path|redis_url" crate/lib/sinex-satellite-sdk/src/ --type rust | grep -v "figment_config.rs" | head -20Repository: Sinity/sinex
Length of output: 239
🏁 Script executed:
rg -n "TODO|FIXME|deprecated|legacy" crate/lib/sinex-satellite-sdk/src/figment_config.rsRepository: Sinity/sinex
Length of output: 38
SatelliteFigmentConfig contains socket_path and redis_url fields that differ from SatelliteConfig.
SatelliteFigmentConfig in figment_config.rs defines socket_path (default: /tmp/sinex-ingestd.sock) and redis_url (default: redis://localhost:6379), while the newer SatelliteConfig in config.rs uses nats_url instead. This architectural inconsistency—where SatelliteFigmentConfig targets gRPC+Redis patterns while SatelliteConfig targets NATS—may confuse users about which configuration structure to use. The socket_path is actively used in sensd_client.rs, but redis_url appears unused outside its definition.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@crate/lib/sinex-satellite-sdk/src/config.rs` around lines 328 - 330,
SatelliteFigmentConfig and SatelliteConfig diverge: SatelliteFigmentConfig
defines socket_path and redis_url (default /tmp/sinex-ingestd.sock and
redis://localhost:6379) while SatelliteConfig uses nats_url; align them by
updating SatelliteFigmentConfig in figment_config.rs to include nats_url with
the same default as default_nats_url() (and remove or mark redis_url deprecated
if unused), keep socket_path in place because sensd_client.rs depends on it, and
remove any unused redis_url references to avoid confusion; ensure the struct
name SatelliteFigmentConfig and the field names socket_path, redis_url, and
nats_url are updated consistently and tests/usage updated accordingly.
Comment on lines
+1006
to
+1016
| // Start LeaseManager if configured (for LeaderStandby model) | ||
| if let Some(lease_manager) = &self.lease_manager { | ||
| info!("Starting lease manager for leader election"); | ||
| lease_manager.start().await?; | ||
|
|
||
| // Wait for initial lease acquisition | ||
| tokio::time::sleep(std::time::Duration::from_secs(2)).await; | ||
|
|
||
| let status = lease_manager.status().await; | ||
| info!(?status, "Lease manager started with initial status"); | ||
| } |
There was a problem hiding this comment.
🧹 Nitpick | 🔵 Trivial
Consider replacing fixed sleep with readiness check.
The 2-second sleep for initial lease acquisition is a brittle heuristic. If lease acquisition takes longer under load, the automaton may proceed without leadership.
🛡️ Proposed readiness loop
// Start LeaseManager if configured (for LeaderStandby model)
if let Some(lease_manager) = &self.lease_manager {
info!("Starting lease manager for leader election");
lease_manager.start().await?;
- // Wait for initial lease acquisition
- tokio::time::sleep(std::time::Duration::from_secs(2)).await;
+ // Wait for initial lease status to stabilize
+ for _ in 0..20 {
+ let status = lease_manager.status().await;
+ if status != crate::LeaseStatus::Unknown {
+ break;
+ }
+ tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+ }
let status = lease_manager.status().await;
info!(?status, "Lease manager started with initial status");
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| // Start LeaseManager if configured (for LeaderStandby model) | |
| if let Some(lease_manager) = &self.lease_manager { | |
| info!("Starting lease manager for leader election"); | |
| lease_manager.start().await?; | |
| // Wait for initial lease acquisition | |
| tokio::time::sleep(std::time::Duration::from_secs(2)).await; | |
| let status = lease_manager.status().await; | |
| info!(?status, "Lease manager started with initial status"); | |
| } | |
| // Start LeaseManager if configured (for LeaderStandby model) | |
| if let Some(lease_manager) = &self.lease_manager { | |
| info!("Starting lease manager for leader election"); | |
| lease_manager.start().await?; | |
| // Wait for initial lease status to stabilize | |
| for _ in 0..20 { | |
| let status = lease_manager.status().await; | |
| if status != crate::LeaseStatus::Unknown { | |
| break; | |
| } | |
| tokio::time::sleep(std::time::Duration::from_millis(100)).await; | |
| } | |
| let status = lease_manager.status().await; | |
| info!(?status, "Lease manager started with initial status"); | |
| } |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@crate/lib/sinex-satellite-sdk/src/stream_processor.rs` around lines 1006 -
1016, Replace the fixed 2s sleep after lease_manager.start() with a readiness
loop that polls lease_manager.status().await until it indicates leadership (or a
ready state) or a deadline/timeout is reached; continue to call
lease_manager.start().await then loop calling lease_manager.status().await with
a small exponential/backoff delay (e.g., 100-500ms) and a configurable timeout,
logging interim statuses and failing/returning an error if the timeout elapses,
so that the automaton only proceeds when self.lease_manager reports the initial
lease acquisition.
Comment on lines
+1026
to
+1036
| // Check lease status before processing (for LeaderStandby model) | ||
| if self.processing_model == crate::ProcessingModel::LeaderStandby { | ||
| if let Some(lease_manager) = &self.lease_manager { | ||
| let status = lease_manager.status().await; | ||
| if status != crate::LeaseStatus::Leader { | ||
| info!(?status, "Not leader, skipping processing"); | ||
| return Ok(()); | ||
| } | ||
| info!("Confirmed as leader, proceeding with processing"); | ||
| } | ||
| } |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Verify if LeaseManager has a wait_for_leadership or similar method
ast-grep --pattern 'impl LeaseManager {
$$$
pub async fn $_($_, $_) {
$$$
}
$$$
}'
# Check how standbys are expected to behave
rg -nC5 "LeaderStandby" --type rustRepository: Sinity/sinex
Length of output: 5006
🏁 Script executed:
# Get the full run_automaton_continuous_mode function
sed -n '1003,1050p' crate/lib/sinex-satellite-sdk/src/stream_processor.rsRepository: Sinity/sinex
Length of output: 2198
🏁 Script executed:
# Find all calls to run_automaton_continuous_mode to see the calling context
rg -n "run_automaton_continuous_mode" crate/lib/sinex-satellite-sdk/src/stream_processor.rs -A 3 -B 3Repository: Sinity/sinex
Length of output: 588
🏁 Script executed:
# Check if there's a loop in the automaton startup that repeatedly calls this
rg -n "loop\|while\|continuous" crate/lib/sinex-satellite-sdk/src/stream_processor.rs -A 2 -B 2Repository: Sinity/sinex
Length of output: 38
🏁 Script executed:
# Find where the run() method is called
rg -n "\.run\(\)" crate/lib/sinex-satellite-sdk/src/stream_processor.rs -A 5 -B 5Repository: Sinity/sinex
Length of output: 38
🏁 Script executed:
# Search in the broader codebase for run() calls
rg -n "\.run\(\)" crate/lib/sinex-satellite-sdk/src/ --type rust -A 2Repository: Sinity/sinex
Length of output: 265
🏁 Script executed:
# Look for the LeaseManager implementation and its start() method
rg -n "impl.*LeaseManager\|pub async fn start" crate/lib/sinex-satellite-sdk/src/ --type rust -A 10Repository: Sinity/sinex
Length of output: 38
🏁 Script executed:
# Find LeaseManager definition - it might be imported or in another module
rg -n "LeaseManager" crate/lib/sinex-satellite-sdk/src/ --type rust | grep "struct\|impl\|type" | head -20Repository: Sinity/sinex
Length of output: 387
🏁 Script executed:
# Check event_processor.rs around line 248 to see the loop context
sed -n '240,260p' crate/lib/sinex-satellite-sdk/src/event_processor.rsRepository: Sinity/sinex
Length of output: 474
🏁 Script executed:
# Look for any loop wrapping around the run() call
sed -n '200,260p' crate/lib/sinex-satellite-sdk/src/event_processor.rsRepository: Sinity/sinex
Length of output: 1563
🏁 Script executed:
# Get the full run() method in stream_processor.rs
rg -n "async fn run\(" crate/lib/sinex-satellite-sdk/src/stream_processor.rs -A 50Repository: Sinity/sinex
Length of output: 38
🏁 Script executed:
# Check LeaseManager.start() implementation
sed -n '57,150p' crate/lib/sinex-satellite-sdk/src/lease_manager.rsRepository: Sinity/sinex
Length of output: 2801
🏁 Script executed:
# Look at the full run_automaton_continuous_mode to understand if there's any retry/loop inside
sed -n '1003,1080p' crate/lib/sinex-satellite-sdk/src/stream_processor.rsRepository: Sinity/sinex
Length of output: 3246
🏁 Script executed:
# Find and display the full EventProcessor.run() method
rg -n "impl EventProcessor" crate/lib/sinex-satellite-sdk/src/event_processor.rs -A 200 | head -250Repository: Sinity/sinex
Length of output: 6959
🏁 Script executed:
# Search for any loop or retry mechanism in stream_processor.rs run method
sed -n '920,950p' crate/lib/sinex-satellite-sdk/src/stream_processor.rsRepository: Sinity/sinex
Length of output: 1241
Standby automaton exits immediately instead of waiting for leadership.
When not leader, run_automaton_continuous_mode returns Ok(()) immediately and exits. Although lease_manager.start() spawns a background task that continuously monitors leadership status, the function checks the status only once at initialization. If the instance becomes leader later through failover, the automaton will not resume processing since the function has already exited. For a LeaderStandby model, standbys should loop and monitor the lease status, becoming active upon acquiring leadership.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@crate/lib/sinex-satellite-sdk/src/stream_processor.rs` around lines 1026 -
1036, The function run_automaton_continuous_mode currently checks
lease_manager.status() once and returns early for
ProcessingModel::LeaderStandby, causing the standby to exit instead of waiting;
change the logic in run_automaton_continuous_mode so that when
self.processing_model == crate::ProcessingModel::LeaderStandby and a
lease_manager exists you do not return immediately on non-Leader status but
instead wait for leadership (either by subscribing to a lease status change
event on lease_manager, calling a provided wait_for_leadership-like API, or
implementing a short async loop/poll that rechecks lease_manager.status().await
until it equals crate::LeaseStatus::Leader), only proceeding to start processing
after confirming LeaseStatus::Leader and keeping the function alive to re-check
on loss of leadership (i.e., stop processing and return to waiting if status
changes away from Leader).
Comment on lines
168
to
172
| - Migratory template for each satellite: | ||
| 1. Integrate `AcquisitionManager` + Stage-as-You-Go to capture materials. | ||
| 2. Publish slices/events to JetStream; confirm ingestion locally. | ||
| 2. [x] Publish slices/events to JetStream; confirm ingestion locally. ✅ NatsPublisher integrated | ||
| 3. Delete sensd job submission paths, configs, and CLI switches. | ||
| 4. Ensure recovery (replay from JetStream) and checkpointing works. |
There was a problem hiding this comment.
🧹 Nitpick | 🔵 Trivial
Fix markdown formatting around the numbered list.
Static analysis flagged missing blank lines around headings and fenced code blocks in this area.
🔧 Proposed formatting fix
- Completed: fs-watcher, desktop, terminal, system, document-ingestor, health-aggregator, terminal-canonicalizer
- Blocked: analytics-automaton, search-automaton (deeper refactoring needed)
+
- Migratory template for each satellite:
+
1. Integrate `AcquisitionManager` + Stage-as-You-Go to capture materials.
2. [x] Publish slices/events to JetStream; confirm ingestion locally. ✅ NatsPublisher integrated🧰 Tools
🪛 markdownlint-cli2 (0.22.0)
[warning] 169-169: Headings should be surrounded by blank lines
Expected: 1; Actual: 0; Below
(MD022, blanks-around-headings)
[warning] 170-170: Fenced code blocks should be surrounded by blank lines
(MD031, blanks-around-fences)
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@docs/way.md` around lines 168 - 172, The numbered list around the "Migratory
template for each satellite" block is missing surrounding blank lines and
causing markdown lint errors; edit the section containing the list (mentions of
AcquisitionManager, Stage-as-You-Go, and NatsPublisher) to ensure there's a
blank line before and after the numbered list and any nearby headings or fenced
code blocks, and add/adjust blank lines between list items and adjacent
headings/code fences so the markdown linter no longer reports "missing blank
lines around headings and fenced code blocks."
| - 🚧 **Gateway & CLI**: Operational; iterative improvements | ||
| - 🔨 **AI/LLM Integration**: Early framework and schema; integration ongoing | ||
|
|
||
| ### ✅ Implemented (Working Code) |
There was a problem hiding this comment.
Fix markdown heading spacing to satisfy lint.
There should be a blank line before this heading to comply with MD022.
📝 Suggested markdown fix
- 🔨 **AI/LLM Integration**: Early framework and schema; integration ongoing
+
### ✅ Implemented (Working Code)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| ### ✅ Implemented (Working Code) | |
| - 🔨 **AI/LLM Integration**: Early framework and schema; integration ongoing | |
| ### ✅ Implemented (Working Code) |
🧰 Tools
🪛 markdownlint-cli2 (0.22.0)
[warning] 103-103: Headings should be surrounded by blank lines
Expected: 1; Actual: 0; Below
(MD022, blanks-around-headings)
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@README.md` at line 103, Add a blank line immediately before the markdown
heading "### ✅ Implemented (Working Code)" to satisfy MD022 (ensure there is a
separating empty line above the heading).
There was a problem hiding this comment.
Pull request overview
This PR updates developer-facing documentation and test/runtime plumbing to reflect a NATS JetStream–only ingest pipeline, and adds a JetStream end-to-end integration test path.
Changes:
- Update README and
docs/way.mdto describe the JetStream-first ingestion + confirmation flow and current implementation status. - Remove remaining socket-era fields from test fixtures / SDK-facing config and CLI flags; standardize on
NATS_URL. - Add a JetStream e2e integration test and introduce a small runtime bridge (
AutomatonEventHandler) plus automaton startup wiring for leases/confirmation buffering.
Reviewed changes
Copilot reviewed 15 out of 15 changed files in this pull request and generated 6 comments.
Show a summary per file
| File | Description |
|---|---|
| README.md | Updates the documented “current core flow” to JetStream-only ingest + confirmations. |
| docs/way.md | Adds an implementation status section and updates phase checklists/progress notes. |
| crate/satellites/sinex-search-automaton/src/lib.rs | Adds a compatibility type alias for processor_main!. |
| crate/satellites/sinex-analytics-automaton/src/lib.rs | Adds a compatibility type alias for processor_main!. |
| crate/lib/sinex-test-utils/src/satellite_management_utils.rs | Removes socket-path assumptions and replaces readiness sleep with JetStream readiness check. |
| crate/lib/sinex-satellite-sdk/tests/integration/satellite_lifecycle_test.rs | Updates integration tests to assert nats_url instead of socket/redis fields. |
| crate/lib/sinex-satellite-sdk/tests/integration/satellite_architecture_test.rs | Updates test config creation to use nats_url. |
| crate/lib/sinex-satellite-sdk/src/stream_processor.rs | Wires in LeaseManager + ConfirmationBuffer for automata startup and adds lease gating logic. |
| crate/lib/sinex-satellite-sdk/src/nats_publisher.rs | Exposes the underlying NATS client for downstream wiring. |
| crate/lib/sinex-satellite-sdk/src/lib.rs | Exports the new automaton handler and updates SatelliteArgs to --nats-url/NATS_URL. |
| crate/lib/sinex-satellite-sdk/src/config.rs | Removes socket/redis fields from SatelliteConfig and switches env loading to NATS_URL. |
| crate/lib/sinex-satellite-sdk/src/cli.rs | Makes NATS URL required (defaulted) and removes the legacy gRPC/socket CLI path. |
| crate/lib/sinex-satellite-sdk/src/automaton_event_handler.rs | Adds an adapter implementing ConfirmedEventHandler for the JetStream e2e path. |
| crate/core/sinex-ingestd/tests/jetstream_e2e_integration_test.rs | Adds JetStream end-to-end and idempotency integration tests. |
| Cargo.toml | Notes/keeps experimental automata workspace members disabled. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+1029
to
+1034
| let status = lease_manager.status().await; | ||
| if status != crate::LeaseStatus::Leader { | ||
| info!(?status, "Not leader, skipping processing"); | ||
| return Ok(()); | ||
| } | ||
| info!("Confirmed as leader, proceeding with processing"); |
There was a problem hiding this comment.
LeaderStandby mode currently returns Ok(()) when this instance is not the leader. That makes an automaton service exit immediately (and when wrapped by SatelliteCoordination, it can also cause a tight leader loop / no processing if the DB leader and NATS-lease leader disagree). Instead, keep the service running and periodically await leadership (or block on shutdown), or make lease-based gating mutually exclusive with the existing SatelliteCoordination path so you don't end up with two independent leader elections.
Suggested change
| let status = lease_manager.status().await; | |
| if status != crate::LeaseStatus::Leader { | |
| info!(?status, "Not leader, skipping processing"); | |
| return Ok(()); | |
| } | |
| info!("Confirmed as leader, proceeding with processing"); | |
| loop { | |
| let status = lease_manager.status().await; | |
| if status == crate::LeaseStatus::Leader { | |
| info!("Confirmed as leader, proceeding with processing"); | |
| break; | |
| } | |
| info!(?status, "Not leader, waiting to acquire leadership before processing"); | |
| tokio::time::sleep(std::time::Duration::from_secs(2)).await; | |
| } |
Comment on lines
+1011
to
+1014
| // Wait for initial lease acquisition | ||
| tokio::time::sleep(std::time::Duration::from_secs(2)).await; | ||
|
|
||
| let status = lease_manager.status().await; |
There was a problem hiding this comment.
Using a fixed sleep(2s) to “wait for initial lease acquisition” is racey and can add avoidable startup latency. Prefer polling lease_manager.status() until it transitions away from Unknown (or use an explicit await/notify mechanism in LeaseManager) with a timeout, so startup is deterministic and failures are surfaced clearly.
Suggested change
| // Wait for initial lease acquisition | |
| tokio::time::sleep(std::time::Duration::from_secs(2)).await; | |
| let status = lease_manager.status().await; | |
| // Wait deterministically for the lease manager to leave its initial | |
| // unknown state so startup does not depend on a fixed delay. | |
| let startup_timeout = std::time::Duration::from_secs(10); | |
| let poll_interval = std::time::Duration::from_millis(100); | |
| let started_at = tokio::time::Instant::now(); | |
| let status = loop { | |
| let status = lease_manager.status().await; | |
| if status != crate::LeaseStatus::Unknown { | |
| break status; | |
| } | |
| if started_at.elapsed() >= startup_timeout { | |
| return Err(eyre!( | |
| "lease manager did not acquire an initial status within {:?}", | |
| startup_timeout | |
| ) | |
| .into()); | |
| } | |
| tokio::time::sleep(poll_interval).await; | |
| }; |
Comment on lines
+25
to
+29
| pub struct AutomatonEventHandler { | ||
| /// List of processed event IDs (for verification) | ||
| processed_event_ids: Arc<RwLock<Vec<Ulid>>>, | ||
| /// Counter for processed events | ||
| processed_count: Arc<RwLock<usize>>, |
There was a problem hiding this comment.
AutomatonEventHandler stores every processed event ID in a grow-only Vec. If this adapter is ever used in a long-running automaton (not just tests), memory will grow without bound. Consider making ID tracking optional/test-only (e.g., behind cfg(test) or a feature), or store a bounded ring buffer / only maintain a counter unless IDs are explicitly needed.
Comment on lines
+25
to
+29
| **Satellites**: 6/8 modern and buildable (75% complete) | ||
| - ✅ All using `processor_main!` and NATS JetStream | ||
| - ✅ Zero legacy gRPC patterns | ||
| - ✅ Modern: fs-watcher, desktop, terminal, system, document-ingestor, health-aggregator, terminal-canonicalizer | ||
| - 🔄 Blocked: analytics-automaton, search-automaton (deeper refactoring) |
There was a problem hiding this comment.
The satellite progress counts don’t add up: it says “6/8 modern and buildable”, but the “Modern:” list contains 7 entries, and “Blocked” lists 2 more (which would imply 9 total). Please reconcile the numerator/denominator and lists so the status section is internally consistent.
Comment on lines
+39
to
+62
| // Bootstrap events_raw stream | ||
| let events_raw_stream = env.nats_subject("events_raw"); | ||
| js.get_or_create_stream(jetstream::stream::Config { | ||
| name: events_raw_stream.clone(), | ||
| subjects: vec![env.nats_subject("events.raw.>")], | ||
| retention: jetstream::stream::RetentionPolicy::Limits, | ||
| max_messages: 10_000, | ||
| storage: jetstream::stream::StorageType::File, | ||
| ..Default::default() | ||
| }) | ||
| .await?; | ||
|
|
||
| // Bootstrap events_confirmations stream | ||
| let confirmations_stream = env.nats_subject("events_confirmations"); | ||
| js.get_or_create_stream(jetstream::stream::Config { | ||
| name: confirmations_stream.clone(), | ||
| subjects: vec![env.nats_subject("events.confirmations.>")], | ||
| retention: jetstream::stream::RetentionPolicy::Limits, | ||
| max_messages: 10_000, | ||
| storage: jetstream::stream::StorageType::File, | ||
| ..Default::default() | ||
| }) | ||
| .await?; | ||
|
|
There was a problem hiding this comment.
The test bootstraps events_raw / events_confirmations streams with configs that differ from ingestd’s own JetStreamConsumer::bootstrap_streams (e.g., confirmations stream compaction via max_messages_per_subject = 1, retention/max_age, etc.). This can make the e2e test less representative and can prevent ingestd from applying its intended stream settings once the streams already exist. Prefer letting ingestd bootstrap the streams, or align these get_or_create_stream configs with ingestd’s bootstrap config.
Suggested change
| // Bootstrap events_raw stream | |
| let events_raw_stream = env.nats_subject("events_raw"); | |
| js.get_or_create_stream(jetstream::stream::Config { | |
| name: events_raw_stream.clone(), | |
| subjects: vec![env.nats_subject("events.raw.>")], | |
| retention: jetstream::stream::RetentionPolicy::Limits, | |
| max_messages: 10_000, | |
| storage: jetstream::stream::StorageType::File, | |
| ..Default::default() | |
| }) | |
| .await?; | |
| // Bootstrap events_confirmations stream | |
| let confirmations_stream = env.nats_subject("events_confirmations"); | |
| js.get_or_create_stream(jetstream::stream::Config { | |
| name: confirmations_stream.clone(), | |
| subjects: vec![env.nats_subject("events.confirmations.>")], | |
| retention: jetstream::stream::RetentionPolicy::Limits, | |
| max_messages: 10_000, | |
| storage: jetstream::stream::StorageType::File, | |
| ..Default::default() | |
| }) | |
| .await?; | |
| // Do not pre-create the ingestd-owned streams here. | |
| // Let ingestd bootstrap them so this e2e test exercises the same | |
| // stream configuration that production uses. |
Comment on lines
248
to
250
| log_level: env_var_or_default("SINEX_LOG_LEVEL", default_log_level), | ||
| ingest_socket_path: env_var_or_default("SINEX_INGEST_SOCKET", default_ingest_socket), | ||
| redis_url: env_var_or_default("SINEX_REDIS_URL", default_redis_url), | ||
| nats_url: env_var_or_default("NATS_URL", default_nats_url), | ||
| database_url: std::env::var("DATABASE_URL").ok(), |
There was a problem hiding this comment.
The doc comment for load_from_env still lists SINEX_INGEST_SOCKET / SINEX_REDIS_URL, but the implementation now reads NATS_URL. Please update the documented environment-variable list so it matches the current config surface.
Sinity
added
the
faux-pull-request
Owner
Author
Sinity
added a commit
that referenced
this pull request
Jul 13, 2026
Finish the socket-era cleanup in the public docs and shared test helpers, then
add the runtime bridge needed for a real JetStream end-to-end test.
test-utils:
- remove `socket_path` from `crate/lib/sinex-test-utils/src/satellite_management_utils.rs` and wait for NATS readiness instead of sleeping, so ingestd test fixtures describe the transport that actually exists
- keep `TestIngestdHandle` focused on stream metadata and in-process service control rather than carrying obsolete socket-era fields
sinex-satellite-sdk:
- deprecate and then remove `ingest_socket_path` from `src/{cli.rs,config.rs,lib.rs}`, replacing it with a required `nats_url` path and deleting the unused Redis config fields
- add `crate/lib/sinex-satellite-sdk/src/automaton_event_handler.rs` and wire `LeaseManager` plus `ConfirmationBuffer` into `StreamProcessorRunner`, giving automata a confirmed-event bridge during startup
- restore the analytics and search automaton aliases and add JetStream end-to-end coverage so the new startup path is exercised from the test suite
docs:
- rewrite `README.md` around the NATS-only ingest flow and update `docs/way.md` so the migration phases and remaining JetStream work match the code
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
I cleaned the last socket-era assumptions out of the shared test fixtures and SDK-facing configuration, then added the runtime bridge that makes an actual JetStream end-to-end test possible. The visible doc change is that the README and
docs/way.mdnow describe a NATS-only ingest flow instead of hedging around sockets and gRPC. The more important code change is thatStreamProcessorRunnernow starts automata with lease-election and confirmation-buffer state, andAutomatonEventHandlergives the runtime a simple confirmed-event sink for the e2e path. Together, those changes let the docs stop lying about the transport while also giving the test suite a way to prove the new path. This is still a transitional runtime bridge, not the final streaming automaton architecture, and I call that out in the code and docs.Motivation
After the transport removal, a surprising amount of public surface still looked like the old world.
TestIngestdConfigstill had a socket path.ProcessorCli,SatelliteConfig, andSatelliteArgsstill exposedingest_socket_path.SatelliteConfigalso still carried Redis settings that were no longer part of the ingest path. Those fields were now worse than dead code: they taught the wrong mental model to anyone reading the SDK or writing tests.At the same time, the runtime side still had a gap. We had JetStream consumers, confirmation buffers, and leases, but the automaton startup path did not yet bridge those pieces into something an end-to-end test could observe. I did not want to wait for the full streaming automaton refactor before adding coverage. A small adapter was enough to prove the path and keep the migration honest.
Changes grouped by concern
1. Delete the last socket-shaped fixture fields
crate/lib/sinex-test-utils/src/satellite_management_utils.rsis the first cleanup point.TestIngestdConfigandTestIngestdHandlestop pretending ingestd is configured around a socket path in the test harness, and startup now waits for NATS readiness instead of a fixed sleep. That matters for correctness more than aesthetics: a readiness check expresses the real dependency, while a sleep only papers over timing.This is the same cleanup in spirit as the SDK config work. Once the transport is NATS-only, any fixture field that says otherwise is an attractive nuisance.
2. Make the SDK config match the live transport
The SDK cleanup happens in two passes:
The resulting shape is the one I wanted from the start:
ProcessorCli,SatelliteArgs, andSatelliteConfignow point atNATS_URLdirectly, with a default ofnats://localhost:4222. That makes the public surface match the actual bus and removes a lot of "which of these fields still matters?" noise from the SDK.3. Bridge automata into the confirmation/lease path
The runtime piece is intentionally small and explicit.
AutomatonEventHandlerimplementsConfirmedEventHandlerand keeps enough state to verify that confirmed events are reaching automata startup. I then threadLeaseManagerandConfirmationBufferintoStreamProcessorRunnersetup so automata are launched with the coordination primitives they need in the JetStream path.I am not claiming this is the final streaming automaton interface. The adapter says otherwise in its own module docs. What it does give us is a real bridge between the startup path and the JetStream primitives that already exist, which is enough to write meaningful end-to-end coverage now.
4. Add the e2e proof and line the docs up behind it
The new test is
crate/core/sinex-ingestd/tests/jetstream_e2e_integration_test.rs. It gives the branch a concrete answer to the question "does the NATS-only path actually work end to end?" I also restored the analytics and search automaton aliases so the workspace still has the expected entrypoints for that path.On the docs side, I rewrote the ingest flow in the README and updated
docs/way.mdso the phase/status language reflects what is now complete and what is still a follow-up. The point here is not to produce better marketing text. It is to stop telling developers and reviewers to think in terms of socket-era ingest surfaces that the code no longer uses.Testing
The branch is covered in three places:
crate/core/sinex-ingestd/tests/jetstream_e2e_integration_test.rsStreamProcessorRunnerstartup pathsatellite_management_utils.rsI also relied on the restored automaton aliases to keep the workspace-level entrypoints consistent while the e2e test exercises the new path.
Notes
AutomatonEventHandleris intentionally a bridge. It tracks confirmed events for verification and startup wiring, but it is not the long-term replacement for integratingStatefulStreamProcessorwith streaming consumption. The code comments anddocs/way.mdboth call that out. I also kept the doc and config cleanup in the same branch because removing the stale fields without updating the public narrative would have left the repository more confusing, not less.Ref #91
Summary by CodeRabbit
Release Notes
New Features
Configuration Changes
Testing
Documentation