Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 14 additions & 1 deletion .github/workflows/ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@ jobs:
uses: dtolnay/rust-toolchain@stable
with:
components: rustfmt, clippy
targets: wasm32-unknown-unknown

- name: Cache cargo
uses: Swatinem/rust-cache@v2
Expand All @@ -43,12 +44,25 @@ jobs:
- name: Run tests
run: cargo test --all --verbose

- name: Compile Rust WebAssembly UDF examples
run: >-
cargo check
--manifest-path udf-sdk/rust/Cargo.toml
--workspace
--target wasm32-unknown-unknown
--locked

- name: Verify checked-in WebAssembly UDF fixtures
working-directory: tests/fixtures/wasm/generated
run: sha256sum -c SHA256SUMS

- name: Validate promoted example configs
run: |
cargo run --quiet --bin streamforge-validate -- examples/configs/config.example.yaml
cargo run --quiet --bin streamforge-validate -- examples/redpanda/selective-replication.yaml
cargo run --quiet --bin streamforge-validate -- examples/production/pii-redaction.yaml
cargo run --quiet --bin streamforge-validate -- examples/production/cdc-to-datalake.yaml
cargo run --quiet --bin streamforge-validate -- tests/fixtures/wasm/release-smoke.yaml

- name: Run clippy
run: cargo clippy --all-targets --all-features -- -D warnings
Expand Down Expand Up @@ -99,7 +113,6 @@ jobs:
rust-security:
name: Rust - Security Audit
runs-on: ubuntu-latest
continue-on-error: true # Don't fail CI on advisories
steps:
- uses: actions/checkout@v4

Expand Down
16 changes: 15 additions & 1 deletion .github/workflows/performance-test.yml
Original file line number Diff line number Diff line change
Expand Up @@ -60,28 +60,41 @@ jobs:
- name: Prepare structured result directories
run: mkdir -p target/performance-results/criterion

- name: Verify checked-in WASM benchmark fixtures
working-directory: tests/fixtures/wasm/generated
run: shasum -a 256 -c SHA256SUMS

- name: Run all Criterion targets
run: |
FILTER_LOG=target/performance-results/criterion/filter_benchmarks.log
TRANSFORM_LOG=target/performance-results/criterion/transform_benchmarks.log
PIPELINE_LOG=target/performance-results/criterion/end_to_end_benchmark.log
WASM_LOG=target/performance-results/criterion/wasm_udf_benchmarks.log
cargo bench --bench filter_benchmarks -- --noplot \
2>&1 | tee "$FILTER_LOG"
cargo bench --bench transform_benchmarks -- --noplot \
2>&1 | tee "$TRANSFORM_LOG"
cargo bench --bench end_to_end_benchmark -- --noplot \
2>&1 | tee "$PIPELINE_LOG"
cargo bench --bench wasm_udf_benchmarks -- --noplot \
2>&1 | tee "$WASM_LOG"

- name: Explain evidence scope
if: always()
run: |
{
echo "## Criterion smoke evidence"
echo
echo "All three repository Criterion targets were executed" \
echo "All four repository Criterion targets were executed" \
"on a shared GitHub-hosted runner."
echo "Results are diagnostic smoke evidence only;" \
"this workflow applies no regression threshold."
echo
echo "The stable wasm_udf/native_no_udf control preserves the" \
"JSON-bytes-to-one-destination-envelope workload."
echo "The <=3% native no-UDF release budget requires statistically" \
"significant matched A/B runs on controlled native hardware;" \
"shared-runner results must not be used as that gate."
} >> "$GITHUB_STEP_SUMMARY"

- name: Upload Criterion reports and logs
Expand All @@ -94,6 +107,7 @@ jobs:
path: |
target/criterion/
target/performance-results/criterion/
tests/fixtures/wasm/generated/SHA256SUMS

throughput-smoke:
name: Throughput smoke evidence
Expand Down
5 changes: 5 additions & 0 deletions .github/workflows/release.yml
Original file line number Diff line number Diff line change
Expand Up @@ -112,6 +112,11 @@ jobs:
- name: Build release binary
run: cargo build --release --locked --target ${{ matrix.target }}

- name: Smoke-test WebAssembly UDF startup
run: >-
target/${{ matrix.target }}/release/streamforge-validate
tests/fixtures/wasm/release-smoke.yaml

- name: Strip binary (Linux only)
if: runner.os == 'Linux'
run: strip target/${{ matrix.target }}/release/streamforge
Expand Down
39 changes: 38 additions & 1 deletion ARCHITECTURE.md
Original file line number Diff line number Diff line change
Expand Up @@ -119,6 +119,43 @@ Function-style array `any` and `all` evaluation still clone each visited array
element into a temporary envelope. That boundary is intentionally left for a
later measured refactor.

### Optional WebAssembly UDF runtime

`src/wasm/` implements the opt-in, stateless UDF extension point. It does not
replace the native DSL and is not initialized when the top-level `wasm`
registry is absent.

Startup canonicalizes the configured artifact root, reads each component once
through bounded I/O, verifies its pinned SHA-256 digest, compiles it with
Wasmtime, links it against an empty host linker, checks its declared WIT world,
and probes instantiation before any Kafka client is created. No WASI or other
ambient host interface is linked.

The versioned `streamforge:udf@1.0.0` WIT package defines separate filter,
JSON-value-transform, and mutable-envelope-transform worlds. Source
topic/partition/offset are input-only. Native and UDF stages compose in this
order:

1. native filter, then UDF filter;
2. native value transform, then UDF value transform;
3. native key/header/timestamp envelope mutations, then UDF envelope mutation.

This ordering implements the `PROJECT_SPEC.md` contract: envelope mutations
observe the final destination payload. It changes the earlier runtime behavior,
which applied native envelope mutations before the value transform. Pipelines
that derive envelope fields from values removed by their value transform must
retain those inputs in the transformed payload or update the envelope rule.

Each invocation uses a fresh store and component instance backed by Wasmtime's
pooling allocator. A dedicated epoch thread enforces execution deadlines.
Configured bounds cover artifact, input, output, linear memory, tables, stack,
and concurrent instances. Guest state is never a persistence contract.

UDF failures are deterministic destination-stage failures and are not retried.
The destination `error_policy` selects fail-fast, one contextual DLQ record,
destination skip, or unchanged-envelope continuation. See `docs/WASM_UDFS.md`
for the ABI, deployment, security, and performance contract.

### Destination processing

`src/processor.rs` builds a runtime for each configured destination.
Expand Down Expand Up @@ -301,4 +338,4 @@ and are not inferred from unit or microbenchmark success.
- `docs/PERFORMANCE.md` — tuning and benchmark method
- `docs/DELIVERY_GUARANTEES.md` — commit and failure semantics

**Last updated:** 2026-07-24
**Last updated:** 2026-07-25
Loading
Loading