faucet-source-mssql-cdc 1.0.0

Microsoft SQL Server CDC (change data capture) source for the faucet-stream ecosystem
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127

# faucet-source-mssql-cdc

Microsoft SQL Server **CDC (change data capture)** source for the
[`faucet-stream`](https://crates.io/crates/faucet-stream) ecosystem.

It polls native SQL Server change data capture — `sys.fn_cdc_get_max_lsn()` for
the high-water LSN, then `cdc.fn_cdc_get_all_changes_<capture_instance>()` per
configured capture instance — and emits per-row change events as CDC envelopes.
Progress is a durable, per-capture-instance **LSN bookmark**, so a resumed run
never re-reads an already-committed change.

## Highlights

- **Native CDC, LSN-driven.** Advances by Log Sequence Number using the built-in
  `cdc.fn_cdc_get_all_changes_*` table-valued functions — no triggers, no
  polling of the base table.
- **Per-transaction durability.** Change rows are buffered by commit LSN
  (`__$start_lsn`); each committed transaction is emitted as its own
  `StreamPage` with a bookmark, so nothing partially-visible ever leaks.
- **Resumable & exactly-once capable.** The bookmark is a monotonic LSN; on
  resume the next poll starts at `increment(bookmark)`. `supports_exactly_once()`
  is `true`, and `capture_resume_position()` anchors CDC before a bulk snapshot
  for `faucet replicate`.
- **Preflight checks.** `new()` (and `faucet doctor`) verify CDC is enabled on
  the database and that every configured capture instance exists, with
  actionable errors.

## Configuration

```yaml
pipeline:
  source:
    type: mssql-cdc
    config:
      # faucet-common-mssql connection block (connection_url OR connection_string + tls)
      connection_url: "mssql://faucet:${env:MSSQL_PASSWORD}@sqlserver:1433/sales"
      tls: { type: trust_server_certificate }

      capture_instances: ["dbo_Orders", "dbo_Items"]  # required, non-empty
      start_position: { type: current }                # current (default) | earliest
      poll_interval: 1                                 # seconds between empty polls
      idle_timeout: 30                                 # end the fetch cycle after this much quiet
      batch_size: 1000                                 # 0 = one aggregate page (test/snapshot)
      max_staged_records: 500000                       # cap one in-progress transaction
      max_connections: 5
      statement_timeout_secs: 300
      # state_key: "mssql-cdc:sales:dbo_Orders"        # optional override
```

| Field | Default | Notes |
|-------|---------|-------|
| `capture_instances` | — (required) | Capture-instance names (`sys.sp_cdc_enable_table` defaults to `<schema>_<table>`). Only `[A-Za-z0-9_]` accepted (injected into the function name). |
| `start_position` | `current` | `current` skips existing history; `earliest` replays whatever the CDC cleanup job still retains. Ignored once a bookmark exists. |
| `poll_interval` | `1s` | Wait between empty polls. |
| `idle_timeout` | `30s` | End the fetch cycle after this much continuous quiet. A long-running runtime (`faucet schedule` / `faucet serve`) re-invokes to keep tailing. |
| `batch_size` | `1000` | `0` accumulates every change into a single trailing page. |
| `max_staged_records` | unbounded | Abort (typed error) if one in-progress transaction buffers more than this. |
| `state_key` | derived | `mssql-cdc:<db>:<capture_instance>` for one instance, else `mssql-cdc:<db>:<digest>`. |

### Prerequisites

Change data capture must be enabled on the server (requires sysadmin, and SQL
Server Agent running for the capture job to populate the change tables):

```sql
EXEC sys.sp_cdc_enable_db;
EXEC sys.sp_cdc_enable_table
    @source_schema = N'dbo',
    @source_name   = N'Orders',
    @role_name     = NULL,
    @capture_instance = N'dbo_Orders';
```

## Change envelope

Each row becomes a CDC envelope with a normalized `op` marker mapped from
`__$operation`:

| `__$operation` | meaning | `op` |
|----------------|---------|------|
| 1 | delete | `d` |
| 2 | insert | `i` |
| 3 | update (before image) | *skipped* |
| 4 | update (after image) | `u` |

```json
{
  "op": "i",
  "schema": "dbo",
  "table": "Orders",
  "before": null,
  "after": { "id": 1, "amount": "9.99" },
  "lsn": "0000002a000000550003",
  "seqval": "0000002a000000550003"
}
```

Deletes carry the removed row in `before`; inserts and updates carry the current
image in `after` (the source queries in `N'all'` mode, so an update's pre-image
is not emitted).

## Mirroring to an upsert sink

Pair the envelope with the `cdc_unwrap` transform stage (which normalizes `op`
into `__op: "u"`/`"d"` and flattens `after`/`before`) and an upsert-capable sink
to build a live mirror:

```yaml
pipeline:
  transforms:
    - type: cdc_unwrap
  source:
    type: mssql-cdc
    config: { connection_url: "...", capture_instances: ["dbo_Orders"] }
  sink:
    type: postgres
    config:
      connection_url: "..."
      table: orders
      write_mode: upsert
      key: ["id"]
      delete_marker: { field: "__op", values: ["d"] }
```

## License

Licensed under either of Apache-2.0 or MIT at your option.