Hypothesis
We hypothesize that GenStage-based pipeline architectures with dynamic backpressure control can sustain throughput above 50,000 events per second while maintaining end-to-end latency below 100ms at p99, and that hybrid batch-stream processing reduces total processing cost by 40% compared to pure real-time architectures when applied to the Prismatic Platform's intelligence workloads.
Background
The Prismatic Platform ingests data from 250+ OSINT providers, security feeds, compliance databases, and internal telemetry sources. This data flows through multiple transformation stages before reaching storage backends and serving layers. The current architecture uses a combination of GenStage pipelines for real-time processing and scheduled batch jobs for historical analysis.
As the platform scaled from 20 to 90 umbrella applications, pipeline complexity grew non-linearly. Cross-domain data dependencies introduced ordering constraints that the original architecture did not anticipate. The Sparkline contract system added formal interface requirements between pipeline stages, but performance characteristics remained empirically unvalidated.
Prior work in the Broadway ecosystem provided foundational patterns for acknowledgment-based processing, but the Prismatic Platform's requirement for epistemic provenance tracking (NABLA axiom 7: Provenance Mandatory) adds overhead not present in standard Broadway deployments.
This experiment systematically evaluates pipeline architectures across three dimensions: throughput capacity, latency characteristics, and cost efficiency (measured as CPU-seconds per million events processed).
Methodology
We constructed three pipeline variants and evaluated each against identical workloads:
Variant A: Pure Real-Time (GenStage) -- All events processed immediately through a 5-stage GenStage pipeline with demand-driven backpressure. Each stage runs as a pool of 10 consumer processes.
Variant B: Pure Batch -- Events buffered in ETS tables and processed in 5-second micro-batches. Each batch is processed as a single MapReduce operation across 20 worker processes.
Variant C: Hybrid Adaptive -- A control plane monitors event arrival rate and dynamically switches between real-time processing (for rates below 10,000 events/second) and micro-batch processing (for higher rates). The switching threshold is configurable with hysteresis to prevent oscillation.
Each variant was tested with synthetic workloads at 1K, 10K, 50K, and 100K events per second for sustained periods of 30 minutes.
Setup
The hybrid pipeline controller implements the adaptive switching logic:
defmodule PrismaticPipeline.HybridController do
use GenServer
@switch_threshold_high 10_000
@switch_threshold_low 7_500
@measurement_window_ms 5_000
defstruct [
:mode,
:event_count,
:window_start,
:realtime_pipeline,
:batch_pipeline
]
def start_link(opts) do
GenServer.start_link(__MODULE__, opts, name: __MODULE__)
end
@impl true
def init(_opts) do
{:ok, realtime} = PrismaticPipeline.Realtime.start_link([])
{:ok, batch} = PrismaticPipeline.Batch.start_link([])
state = %__MODULE__{
mode: :realtime,
event_count: 0,
window_start: System.monotonic_time(:millisecond),
realtime_pipeline: realtime,
batch_pipeline: batch
}
schedule_rate_check()
{:ok, state}
end
@impl true
def handle_info(:check_rate, state) do
now = System.monotonic_time(:millisecond)
elapsed = now - state.window_start
rate = state.event_count / max(elapsed / 1_000, 0.001)
new_mode = determine_mode(state.mode, rate)
if new_mode != state.mode do
:telemetry.execute(
[:prismatic_pipeline, :mode_switch],
%{rate: rate},
%{from: state.mode, to: new_mode}
)
end
schedule_rate_check()
{:noreply, %{state | mode: new_mode, event_count: 0, window_start: now}}
end
defp determine_mode(:realtime, rate) when rate > @switch_threshold_high, do: :batch
defp determine_mode(:batch, rate) when rate < @switch_threshold_low, do: :realtime
defp determine_mode(current, _rate), do: current
defp schedule_rate_check do
Process.send_after(self(), :check_rate, @measurement_window_ms)
end
endThe provenance-tracking stage adds NABLA compliance to every event:
defmodule PrismaticPipeline.ProvenanceStage do
use GenStage
@impl true
def handle_events(events, _from, state) do
enriched =
Enum.map(events, fn event ->
%{
event
| provenance: %{
source: event.source_id,
timestamp: DateTime.utc_now(),
pipeline_stage: __MODULE__,
transformation_hash: :crypto.hash(:sha256, :erlang.term_to_binary(event.data)),
axiom_compliance: validate_axioms(event)
}
}
end)
{:noreply, enriched, state}
end
endResults
Throughput measurements across all three variants:
| Target Rate | Variant A (Realtime) | Variant B (Batch) | Variant C (Hybrid) |
|---|---|---|---|
| 1K evt/s | 1,000 (100%) | 1,000 (100%) | 1,000 (100%) |
| 10K evt/s | 10,000 (100%) | 10,000 (100%) | 10,000 (100%) |
| 50K evt/s | 47,200 (94.4%) | 50,000 (100%) | 50,000 (100%) |
| 100K evt/s | 61,300 (61.3%) | 98,400 (98.4%) | 94,700 (94.7%) |
Latency at 50K events/second (the target operating point):
| Percentile | Variant A | Variant B | Variant C |
|---|---|---|---|
| p50 | 4.2 ms | 2,510 ms | 5.1 ms |
| p95 | 18.7 ms | 5,020 ms | 22.3 ms |
| p99 | 87.3 ms | 5,040 ms | 89.1 ms |
| p99.9 | 312 ms | 5,050 ms | 142 ms |
Cost efficiency (CPU-seconds per million events):
| Rate | Variant A | Variant B | Variant C |
|---|---|---|---|
| 10K | 42.1 | 28.3 | 38.7 |
| 50K | 89.4 | 31.2 | 52.8 |
| 100K | 184.7 | 33.1 | 58.3 |
Analysis
The results partially confirm our hypothesis. The Hybrid variant achieved the 50K events/second throughput target at p99 latency of 89.1ms, narrowly under the 100ms threshold. However, the cost reduction was 41% compared to pure real-time (Variant A), matching our 40% prediction almost exactly.
The most interesting finding was the non-linear degradation of Variant A above 50K events/second. GenStage backpressure mechanisms worked correctly up to the saturation point, but beyond it, message queue depth grew exponentially, causing cascading latency spikes. The Hybrid variant avoided this by switching to batch mode at high rates, sacrificing individual event latency for aggregate throughput stability.
Variant B's latency characteristics make it unsuitable for real-time security alerting (the primary use case for Prismatic Perimeter), but its cost efficiency at high rates is 3.2x better than real-time processing. This validates batch processing for historical analysis workloads.
The provenance tracking overhead added approximately 1.2ms per event at p50 and 4.8ms at p99. This is acceptable for NABLA compliance but represents 25% of the total p50 latency budget, suggesting optimization opportunities in the hashing step.
Conclusions
- Hybrid pipelines are the correct architecture for workloads spanning security alerting and historical analysis.
- GenStage backpressure is insufficient at sustained rates above 50K events/second without architectural complementation.
- Provenance tracking is viable but requires hash optimization to reduce its 25% latency share.
- Batch processing cost efficiency makes it the clear choice for non-latency-sensitive workloads.
- Hysteresis in mode switching is essential to prevent oscillation at boundary rates.
Next Steps
- Optimize provenance hashing with NIF-based SHA-256 for 3x improvement
- Test Variant C with real production traffic patterns (not synthetic)
- Evaluate Broadway integration for acknowledgment-based processing
- Extend the hybrid controller with predictive rate estimation using EWMA
- Benchmark cross-datacenter pipeline replication latency
Related Experiments
- Storage Benchmarks -- Backend performance that pipelines write to
- OSINT Pipeline -- Real-world pipeline workloads from intelligence sources
- Drift Detection -- Detecting pipeline behavior changes over time
- Multi-Agent Coordination -- Agent-driven pipeline orchestration
Connect & Contribute
Created by Tomáš Korcak (korczis) | Open Source under GHL