> ## Documentation Index
> Fetch the complete documentation index at: https://ventstream.dev/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# Testing a pipeline

> Verify create / update / delete propagation and source↔index consistency.

Once an agent is tailing, you want confidence that every kind of change
reaches the index correctly. This page is the test recipe used to
validate the engine — adapt it to your own projection.

## Automated integration suite

The repository includes Docker-backed integration suites under
`crates/ventstream/tests/`. They start isolated source and OpenSearch
containers and drive the engine end to end. Coverage includes bootstrap, live
changes, deletes, cursor restart, join-state recovery, and source-specific
failure handling.

These tests are marked `#[ignore]` so the fast unit run skips them; run
one source or the complete sequential matrix explicitly:

```bash theme={null}
./scripts/test-sources.sh postgres
./scripts/test-sources.sh neo4j
./scripts/test-sources.sh mongodb
./scripts/test-sources.sh mysql
./scripts/test-sources.sh kafka

# Run every source sequentially to bound local memory use.
./scripts/test-sources.sh all
```

These are local qualification tests, not part of the default pull-request
suite. Run the source relevant to your change before release.

The Redis sink has a separate local qualification runner. It provisions
isolated standalone, authenticated, RedisJSON, replica, failover, pressure,
mutual-TLS, Sentinel, and Cluster deployments, then exercises recovery and
topology changes serially:

```bash theme={null}
./scripts/test-redis-sink.sh all
```

Use `contracts` or `topology` instead of `all` to run one half of the suite.
Set `VS_TEST_REDIS_KEEP=1` only when you need to retain the containers after a
failure for inspection.

## The matrix

For each path your spec embeds, exercise all three operations and
confirm the document converges:

| Operation  | What to check                                          |
| ---------- | ------------------------------------------------------ |
| **CREATE** | a new related row/edge → the embedded field appears    |
| **UPDATE** | change an embedded value → the document reflects it    |
| **DELETE** | remove the row/edge → the field clears / array shrinks |

## The sibling check (the important one)

For any **shared** endpoint (a customer many orders reference, a category
many products reference), do the change on one primary and assert a
**sibling** primary that shares the endpoint is **undisturbed**.

This is the test that proves the [fan-out](/docs/concepts/fan-out) is exactly
right — it catches both failure modes:

* **over-propagation** — the sibling wrongly changed (hot-endpoint bug)
* **under-propagation** — the target didn't change (missed fan-out)

Run this check for every shared endpoint in the projection. A property change on
a shared node is different: it should update every document that embeds that
property.

## Latency expectations

Measure source commit to target visibility under representative load. The result
depends on source polling, dispatch flush settings, projection depth, fan-out,
and target bulk latency. Record both steady-state latency and the time required
to drain a bounded burst.

## Consistency check

After a batch of changes, confirm counts match:

```bash theme={null}
# Neo4j
NEO4J_AUTH="neo4j:${VS_NEO4J_PASSWORD:?set VS_NEO4J_PASSWORD}"
curl -s --user "$NEO4J_AUTH" \
  -X POST http://localhost:7474/db/neo4j/tx/commit \
  -H 'Content-Type: application/json' \
  -d '{"statements":[{"statement":"MATCH (p:Product) RETURN count(p)"}]}'

# OpenSearch
curl -s 'http://localhost:9200/products_denormalized/_count'
```

They should be equal once the engine has drained its queue. Check source lag and
engine metrics/logs to confirm the pipeline is idle before comparing.

## Burst testing

To check the engine holds up under load, fire a mixed batch (e.g. 500
updates + 250 deletes + 250 creates) in one transaction and watch:

* **CDC lag** drains to zero (no growing backlog)
* **RSS** stays bounded (no per-event accumulation)
* **count parity** holds afterward

Record the drain time, peak RSS, retry count, and final source-to-target parity.

## Drain-resume + reconciliation

To verify delete reconciliation:

<Steps>
  <Step title="Note the index count and pick rows">Record the baseline.</Step>
  <Step title="Drain the pipeline">Use `ventstreamctl pipelines drain ... --wait` and verify the durable operation succeeds.</Step>
  <Step title="Delete rows in the source while drained">The agent isn't watching.</Step>
  <Step title="Rebootstrap and resume">Run the explicit rebootstrap operation, then resume and wait for both operations.</Step>
  <Step title="Assert">The deleted rows' docs are gone AND counts match.</Step>
</Steps>

## What won't propagate (and shouldn't)

Updating a node/table the spec never traverses produces zero
recomputations. Verify this as a negative test so an unrelated source change
cannot create unexpected target writes.