Noundry.Streams

Overview

Noundry Streams is a high-performance event streaming service that ingests, persists, and relays messages. It uses System.Threading.Channels for lock-free queuing and Tuxedo ORM for data access. The service targets .NET 8.0, 9.0, and 10.0 (multi-target) and uses ASP.NET Core Minimal APIs.

Ingest

HTTP message ingestion with sub-millisecond latency. Content-agnostic payload handling.

Persist

Crash-safe SQLite with WAL mode. Batch writing with configurable intervals.

Relay

HTTP endpoint forwarding with configurable retry logic, backoff, and dead letter queue.

Architecture

src/
NoundryStreams.Core/ # Core library (no ASP.NET dependency)
 Configuration/ # NoundryStreamsOptions
 Database/ # Repositories (Tuxedo ORM)
 Models/ # StreamConfig, RelayConfig, Messages
 Services/ # StreamManager, BatchWriter, RelayWorker
 Telemetry/ # OpenTelemetry integration

NoundryStreams.Service/ # ASP.NET Core host
 Endpoints/ # IngestEndpoints, AdminEndpoints
 Middleware/ # ApiKeyAuth, ExceptionHandling

Core Models

StreamConfig

public sealed class StreamConfig
{
 public required string StreamId { get; init; } // Unique identifier
 public string? DisplayName { get; init; }
 public bool OrderingRequired { get; init; } // Sequential processing
 public int RetentionHours { get; init; } = 168; // 7 days default
 public string? OtelEndpoint { get; init; } // OpenTelemetry OTLP/gRPC
 public int ChannelCapacity { get; init; } = 5000; // Bounded channel (backpressure)
 public int BatchIntervalMs { get; init; } = 50; // Max wait before batch write
 public int BatchMaxSize { get; init; } = 500; // Max messages per transaction
 public int RelayWorkerCount { get; init; } = 4; // Parallel workers
 public bool Enabled { get; init; } = true;
}

RelayConfig

public sealed class RelayConfig
{
 public required string StreamId { get; init; }
 public required string EndpointUrl { get; init; } // HTTP/HTTPS destination
 public string HttpMethod { get; init; } = "POST";
 public string? Headers { get; init; } // JSON: {"Authorization": "Bearer x"}
 public int TimeoutMs { get; init; } = 5000;
 public int RetryAttempts { get; init; } = 3;
 public int RetryBackoffMs { get; init; } = 1000; // Base: attempt * backoff
}

IncomingMessage

public sealed record IncomingMessage
{
 public required string StreamId { get; init; }
 public required string ContentType { get; init; } // MIME type from HTTP header
 public required byte[] Payload { get; init; } // Raw bytes (content-agnostic)
 public required long ReceivedAt { get; init; } // Unix timestamp ms
 public bool OrderingRequired { get; init; } // Per-message ordering override
}

HTTP API Reference

Ingestion Endpoints

POST /streams/{streamId}/ingest

Primary message ingestion endpoint. Accepts raw bytes with any content type.

Success: 202 Accepted | Errors: 404 (stream not found), 400 (disabled), 503 (backpressure)

# Example ingestion request
$ curl -X POST http://localhost:5000/streams/webhook-events/ingest \
 -H "Content-Type: application/json" \
 -H "X-API-Key: your-api-key" \
 -d '{"event": "user.signup", "userId": 12345}'

POST /ingest

Alternative endpoint with header/query-based stream selection. Priority: X-Stream-Id header > stream query parameter.

Admin Endpoints

Method	Endpoint	Description
POST	/admin/streams	Create a new stream
GET	/admin/streams	List all active streams
GET	/admin/streams/{streamId}	Get stream details
POST	/admin/streams/{streamId}/relays	Add relay configuration
GET	/admin/streams/{streamId}/metrics	Get real-time stream metrics
PUT	/admin/streams/{streamId}/enable	Enable a stream
PUT	/admin/streams/{streamId}/disable	Disable a stream
POST	/admin/streams/{streamId}/cleanup	Trigger immediate cleanup of expired messages
GET	/admin/streams/{streamId}/stats	Get stream statistics
GET	/metrics	Prometheus-compatible metrics for all streams

Create Stream Example

$ curl -X POST http://localhost:5000/admin/streams \
 -H "Content-Type: application/json" \
 -d '{
 "streamId": "webhook-events",
 "displayName": "Webhook Events",
 "retentionHours": 168,
 "channelCapacity": 10000,
 "batchMaxSize": 500,
 "relayWorkerCount": 4
 }'

# Response: 201 Created

Add Relay Endpoint

$ curl -X POST http://localhost:5000/admin/streams/webhook-events/relays \
 -H "Content-Type: application/json" \
 -d '{
 "endpointUrl": "https://api.example.com/webhooks",
 "httpMethod": "POST",
 "headers": {"Authorization": "Bearer token123"},
 "timeoutMs": 5000,
 "retryAttempts": 3,
 "retryBackoffMs": 1000
 }'

# Response: 201 Created {"configId": 1}

Configuration

// appsettings.json
{
 "NoundryStreams": {
 "DataPath": "./data/noundry-streams.db",
 "ApiKey": null,
 "Defaults": {
 "ChannelCapacity": 10000,
 "BatchIntervalMs": 50,
 "BatchMaxSize": 500,
 "RetentionHours": 168,
 "RelayWorkerCount": 4
 },
 "Sqlite": {
 "CacheSizeMB": 256,
 "MmapSizeMB": 1024,
 "WalAutocheckpoint": 1000
 }
 },
 "Kestrel": {
 "Limits": {
 "MaxConcurrentConnections": 100000,
 "MaxRequestBodySize": 10485760
 }
 }
}

Key Components

StreamManager

Orchestrates stream lifecycle. Creates bounded channels, starts BatchWriter and RelayWorker instances per stream. Thread-safe via ConcurrentDictionary.

BatchWriter

Reads from channel, batches messages, writes to SQLite in transactions. Collects until BatchMaxSize OR BatchIntervalMs timeout.

RelayWorker

Picks up persisted messages and forwards to configured HTTP endpoints. Supports parallel or sequential relay with retry and dead letter queue.

RetentionWorker

Background service that runs hourly to delete expired processed messages based on stream retention policy.

SequenceGenerator

Thread-safe monotonic per-stream sequence number generator. Supports atomic batch reservation for high throughput.

StreamTelemetry

OpenTelemetry integration for distributed tracing via OTLP/gRPC. Tracks batch writes and relay operations with stream.id tags.

Message Flow

1 HTTP request arrives at /streams/{id}/ingest -- message written to bounded Channel

2 BatchWriter reads from channel, collects messages until batch is full or timeout elapses

3 Batch INSERT into SQLite with RETURNING clause, sequence numbers assigned atomically

4 Persisted messages forwarded to relay Channel for HTTP delivery

5 RelayWorkers (N parallel) send to configured endpoints with retry logic (attempt * backoff)

6 Failed messages after all retries are written to the dead letter queue for later inspection

Performance

50K+

Messages/second

<1ms

Ingestion latency

WAL

Crash-safe persistence

Powered by System.Threading.Channels for lock-free queuing, SQLite WAL mode for concurrent reads/writes, and configurable batch sizes for throughput optimization.

SQLite Performance Tuning

-- Applied at database initialization
PRAGMA journal_mode = WAL; -- Write-ahead logging
PRAGMA synchronous = NORMAL; -- Balanced durability/performance
PRAGMA cache_size = -262144; -- 256MB cache
PRAGMA temp_store = MEMORY; -- In-memory temp tables
PRAGMA mmap_size = 1073741824; -- 1GB memory-mapped I/O
PRAGMA page_size = 8192; -- Larger pages for throughput
PRAGMA wal_autocheckpoint = 1000; -- Checkpoint every 1000 pages
PRAGMA busy_timeout = 5000; -- 5s wait on lock contention

Stream Metrics

public sealed class StreamMetrics
{
 public long TotalMessagesReceived { get; set; } // Lifetime ingested
 public long TotalMessagesRelayed { get; set; } // Lifetime successful relays
 public long TotalMessagesFailed { get; set; } // Lifetime failed relays
 public double AvgIngestLatencyMs { get; set; }
 public double AvgRelayLatencyMs { get; set; }
 public double ChannelUtilization { get; set; } // 0.0 - 1.0
 public long DiskUsageBytes { get; set; }
}

Prometheus-compatible metrics are also available at the GET /metrics endpoint for integration with Grafana and other monitoring tools.

Health Checks

`GET /health`

Basic health check returning status, timestamp, and version.

`GET /health/ready`

Readiness probe. Returns 200 if database is accessible, 503 otherwise.