-
Notifications
You must be signed in to change notification settings - Fork 59
Expand file tree
/
Copy pathllms-full.txt
More file actions
719 lines (580 loc) · 59.1 KB
/
Copy pathllms-full.txt
File metadata and controls
719 lines (580 loc) · 59.1 KB
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
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
<!-- Code generated by cmd/conduit/internal/llmsgen; DO NOT EDIT. -->
# Conduit
> Broker-neutral data streaming engine and Kafka Connect replacement. Pipelines move records from
> source connectors to destination connectors with processors in between; any-language plugins
> over gRPC/WASM; embeddable; agent-legible with stable error codes and an MCP server.
## 1. Pipeline configuration schema
### Versions
- 2.0
- 2.1
- 2.2 (latest)
### Configuration
| Field | YAML key | Go type | Required |
| --- | --- | --- | --- |
| Version | `version` | `string` | yes |
| Pipelines | `pipelines` | `[]Pipeline` | yes |
### Pipeline
| Field | YAML key | Go type | Required |
| --- | --- | --- | --- |
| ID | `id` | `string` | yes |
| Status | `status` | `string` | yes |
| Name | `name` | `string` | yes |
| Description | `description` | `string` | yes |
| Connectors | `connectors` | `[]Connector` | yes |
| Processors | `processors` | `[]Processor` | yes |
| DLQ | `dead-letter-queue` | `DLQ` | yes |
### Connector
| Field | YAML key | Go type | Required |
| --- | --- | --- | --- |
| ID | `id` | `string` | yes |
| Type | `type` | `string` | yes |
| Plugin | `plugin` | `string` | yes |
| Name | `name` | `string` | yes |
| Settings | `settings` | `map[string]string` | yes |
| Processors | `processors` | `[]Processor` | yes |
### Processor
| Field | YAML key | Go type | Required |
| --- | --- | --- | --- |
| ID | `id` | `string` | yes |
| Type | `type` | `string` | yes |
| Plugin | `plugin` | `string` | yes |
| Condition | `condition` | `string` | yes |
| Settings | `settings` | `map[string]string` | yes |
| Workers | `workers` | `int` | yes |
### DLQ
| Field | YAML key | Go type | Required |
| --- | --- | --- | --- |
| Plugin | `plugin` | `string` | yes |
| Settings | `settings` | `map[string]string` | yes |
| WindowSize | `window-size` | `*int` | no |
| WindowNackThreshold | `window-nack-threshold` | `*int` | no |
### Legacy v1 (1.1)
v1 pipeline configs key `pipelines`, `connectors`, and `processors` by map (`pipelines: {<id>: {...}}`) instead of v2's ordered list, so processor and connector ordering within a v1 file is not deterministic. v1 has no `plugin`/`condition` processor fields and no top-level `dead-letter-queue` before 1.1. New pipelines should use v2.
## 2. Built-in connectors (6)
### file (v0.10.3)
A file source and destination plugin for Conduit.
**How it works**
The Source connector listens for changes appended to the source file and
sends records with the changes.
The Destination connector receives records and writes them to a file.
**Source**
The Source connector only cares to have a valid path, even if the file
doesn't exist, it will still run and wait until a file with the configured
name is there, then it will start listening to changes and sending records.
**Destination**
The Destination connector will create the file if it doesn't exist, and
records with changes will be appended to the destination file when received.
Author: Meroxa, Inc.
#### Source parameters
| Key | Type | Default | Required | Description | Validations |
| --- | --- | --- | --- | --- | --- |
| `path` | string | | yes | Path is the file path used by the connector to read/write records. | required |
| `sdk.batch.delay` | duration | 0 | no | Maximum delay before an incomplete batch is read from the source. | |
| `sdk.batch.size` | int | 0 | no | Maximum size of batch before it gets read from the source. | greater-than(-1) |
| `sdk.schema.context.enabled` | bool | true | no | Specifies whether to use a schema context name. If set to false, no schema context name will be used, and schemas will be saved with the subject name specified in the connector (not safe because of name conflicts). | |
| `sdk.schema.context.name` | string | | no | Schema context name to be used. Used as a prefix for all schema subject names. If empty, defaults to the connector ID. | |
| `sdk.schema.extract.key.enabled` | bool | false | no | Whether to extract and encode the record key with a schema. | |
| `sdk.schema.extract.key.subject` | string | key | no | The subject of the key schema. If the record metadata contains the field "opencdc.collection" it is prepended to the subject name and separated with a dot. | |
| `sdk.schema.extract.payload.enabled` | bool | false | no | Whether to extract and encode the record payload with a schema. | |
| `sdk.schema.extract.payload.subject` | string | payload | no | The subject of the payload schema. If the record metadata contains the field "opencdc.collection" it is prepended to the subject name and separated with a dot. | |
| `sdk.schema.extract.type` | string | avro | no | The type of the payload schema. | inclusion(avro) |
#### Destination parameters
| Key | Type | Default | Required | Description | Validations |
| --- | --- | --- | --- | --- | --- |
| `path` | string | | yes | Path is the file path used by the connector to read/write records. | required |
| `sdk.batch.delay` | duration | 0 | no | Maximum delay before an incomplete batch is written to the destination. | |
| `sdk.batch.size` | int | 0 | no | Maximum size of batch before it gets written to the destination. | greater-than(-1) |
| `sdk.rate.burst` | int | 0 | no | Allow bursts of at most X records (0 or less means that bursts are not limited). Only takes effect if a rate limit per second is set. Note that if `sdk.batch.size` is bigger than `sdk.rate.burst`, the effective batch size will be equal to `sdk.rate.burst`. | greater-than(-1) |
| `sdk.rate.perSecond` | float | 0 | no | Maximum number of records written per second (0 means no rate limit). | greater-than(-1) |
| `sdk.record.format` | string | opencdc/json | no | The format of the output record. See the Conduit documentation for a full list of supported formats (https://conduit.io/docs/using/connectors/configuration-parameters/output-format). | |
| `sdk.record.format.options` | string | | no | Options to configure the chosen output record format. Options are normally key=value pairs separated with comma (e.g. opt1=val2,opt2=val2), except for the `template` record format, where options are a Go template. | |
| `sdk.schema.extract.key.enabled` | bool | true | no | Whether to extract and decode the record key with a schema. | |
| `sdk.schema.extract.payload.enabled` | bool | true | no | Whether to extract and decode the record payload with a schema. | |
### generator (v0.10.4)
A plugin capable of generating dummy records (in different formats).
The generator is capable of generating dummy records (in JSON format). The
connector makes it possible to configure the records' fields, the operation
(whether the record was created, updated or deleted), the rate at which records
are produced, and other properties of the connector.
**Examples**
**Bursts**
The following configuration generates 100 records in bursts of 10 records each,
with a 1-second sleep time between bursts.
> [!NOTE]
> The generator currently has no concept of resuming work. For instance, below
> we have configured it to generate 100 records, but if we restart the pipeline
> (by stopping and starting the pipeline or by restarting Conduit), then it will
> start generating the 100 records from scratch.
```yaml
version: 2.2
pipelines:
- id: example
status: running
connectors:
- id: example
type: source
plugin: generator
settings:
# global settings
rate: 10
recordCount: 100
burst.generateTime: 1s
burst.sleepTime: 1s
# default collection
format.type: structured
format.options.id: int
format.options.name: string
operations: create
```
**Collections**
The following configuration generates records forever with a steady rate of 1000
records per second. Records are generated in the `users` and `orders` collections.
The generated records have a different format, depending on the collection they
belong to.
```yaml
version: 2.2
pipelines:
- id: example
status: running
connectors:
- id: example
type: source
plugin: generator
settings:
# global settings
rate: 1000
# collection "users"
collections.users.format.type: structured
collections.users.format.options.id: int
collections.users.format.options.name: string
collections.users.operations: create
# collection "orders"
collections.orders.format.type: raw
collections.orders.format.options.id: int
collections.orders.format.options.product: string
collections.orders.operations: create,update,delete
```
Author: Meroxa, Inc.
#### Source parameters
| Key | Type | Default | Required | Description | Validations |
| --- | --- | --- | --- | --- | --- |
| `burst.generateTime` | duration | 1s | no | The amount of time the generator is generating records in a burst. Has an effect only if `burst.sleepTime` is set. | |
| `burst.sleepTime` | duration | | no | The time the generator "sleeps" between bursts. | |
| `collections.*.format.options.*` | string | | no | The options for the `raw` and `structured` format types. It accepts pairs of field names and field types, where the type can be one of: `int`, `string`, `time`, `bool`, `duration`. | |
| `collections.*.format.options.path` | string | | no | Path to the input file (only applicable if the format type is `file`). | |
| `collections.*.format.type` | string | | no | The format of the generated payload data (raw, structured, file). | inclusion(raw,structured,file) |
| `collections.*.operations` | string | create | yes | Comma separated list of record operations to generate. Allowed values are "create", "update", "delete", "snapshot". | required |
| `format.options.*` | string | | no | The options for the `raw` and `structured` format types. It accepts pairs of field names and field types, where the type can be one of: `int`, `string`, `time`, `bool`, `duration`. | |
| `format.options.path` | string | | no | Path to the input file (only applicable if the format type is `file`). | |
| `format.type` | string | | no | The format of the generated payload data (raw, structured, file). | inclusion(raw,structured,file) |
| `operations` | string | create | yes | Comma separated list of record operations to generate. Allowed values are "create", "update", "delete", "snapshot". | required |
| `rate` | float | | no | The maximum rate in records per second, at which records are generated (0 means no rate limit). | |
| `readTime` | duration | | no | The time it takes to 'read' a record. Deprecated: use `rate` instead. | |
| `recordCount` | int | | no | Number of records to be generated (0 means infinite). | greater-than(-1) |
| `sdk.batch.delay` | duration | 0 | no | Maximum delay before an incomplete batch is read from the source. | |
| `sdk.batch.size` | int | 0 | no | Maximum size of batch before it gets read from the source. | greater-than(-1) |
| `sdk.schema.context.enabled` | bool | true | no | Specifies whether to use a schema context name. If set to false, no schema context name will be used, and schemas will be saved with the subject name specified in the connector (not safe because of name conflicts). | |
| `sdk.schema.context.name` | string | | no | Schema context name to be used. Used as a prefix for all schema subject names. If empty, defaults to the connector ID. | |
| `sdk.schema.extract.key.enabled` | bool | true | no | Whether to extract and encode the record key with a schema. | |
| `sdk.schema.extract.key.subject` | string | key | no | The subject of the key schema. If the record metadata contains the field "opencdc.collection" it is prepended to the subject name and separated with a dot. | |
| `sdk.schema.extract.payload.enabled` | bool | true | no | Whether to extract and encode the record payload with a schema. | |
| `sdk.schema.extract.payload.subject` | string | payload | no | The subject of the payload schema. If the record metadata contains the field "opencdc.collection" it is prepended to the subject name and separated with a dot. | |
| `sdk.schema.extract.type` | string | avro | no | The type of the payload schema. | inclusion(avro) |
#### Destination parameters
(none)
### kafka (v0.12.3)
A Kafka source and destination plugin for Conduit, written in Go.
**Source**
A Kafka source connector is represented by a single consumer in a Kafka consumer
group. By virtue of that, a source's logical position is the respective
consumer's offset in Kafka. Internally, though, we're not saving the offset as
the position: instead, we're saving the consumer group ID, since that's all
which is needed for Kafka to find the offsets for our consumer.
A source is getting associated with a consumer group ID the first time the
`Read()` method is called.
**Destination**
The destination connector writes records to Kafka.
**Output format**
The output format can be adjusted using configuration options provided by the
connector SDK:
- `sdk.record.format`: used to choose the format
- `sdk.record.format.options`: used to configure the specifics of the chosen format
See [this article](https://conduit.io/docs/connectors/output-formats) for more
info on configuring the output format.
**Batching**
Batching can also be configured using connector SDK provided options:
- `sdk.batch.size`: maximum number of records in batch before it gets written to
the destination (defaults to 0, no batching)
- `sdk.batch.delay`: maximum delay before an incomplete batch is written to the
destination (defaults to 0, no limit)
Author: Meroxa, Inc.
#### Source parameters
| Key | Type | Default | Required | Description | Validations |
| --- | --- | --- | --- | --- | --- |
| `caCert` | string | | no | CACert is the Kafka broker's certificate. | |
| `clientCert` | string | | no | ClientCert is the Kafka client's certificate. | |
| `clientID` | string | conduit-connector-kafka | no | ClientID is a unique identifier for client connections established by this connector. | |
| `clientKey` | string | | no | ClientKey is the Kafka client's private key. | |
| `commitOffsetsDelay` | duration | 5s | no | CommitOffsetsDelay defines how often consumed offsets should be committed. | |
| `commitOffsetsSize` | int | 1000 | no | CommitOffsetsSize defines the maximum number of consumed offsets to be committed at a time. | greater-than(-1) |
| `groupID` | string | | no | GroupID defines the consumer group id. | |
| `insecureSkipVerify` | bool | | no | InsecureSkipVerify defines whether to validate the broker's certificate chain and host name. If 'true', accepts any certificate presented by the server and any host name in that certificate. | |
| `readFromBeginning` | bool | | no | ReadFromBeginning determines from whence the consumer group should begin consuming when it finds a partition without a committed offset. If this options is set to true it will start with the first message in that partition. | |
| `retryGroupJoinErrors` | bool | true | no | RetryGroupJoinErrors determines whether the connector will continually retry on group join errors. | |
| `saslMechanism` | string | | no | Mechanism configures the connector to use SASL authentication. If empty, no authentication will be performed. | inclusion(PLAIN,SCRAM-SHA-256,SCRAM-SHA-512) |
| `saslPassword` | string | | no | Password sets up the password used with SASL authentication. | |
| `saslUsername` | string | | no | Username sets up the username used with SASL authentication. | |
| `sdk.batch.delay` | duration | 0 | no | Maximum delay before an incomplete batch is read from the source. | |
| `sdk.batch.size` | int | 0 | no | Maximum size of batch before it gets read from the source. | greater-than(-1) |
| `sdk.schema.context.enabled` | bool | true | no | Specifies whether to use a schema context name. If set to false, no schema context name will be used, and schemas will be saved with the subject name specified in the connector (not safe because of name conflicts). | |
| `sdk.schema.context.name` | string | | no | Schema context name to be used. Used as a prefix for all schema subject names. If empty, defaults to the connector ID. | |
| `sdk.schema.extract.key.enabled` | bool | false | no | Whether to extract and encode the record key with a schema. | |
| `sdk.schema.extract.key.subject` | string | key | no | The subject of the key schema. If the record metadata contains the field "opencdc.collection" it is prepended to the subject name and separated with a dot. | |
| `sdk.schema.extract.payload.enabled` | bool | false | no | Whether to extract and encode the record payload with a schema. | |
| `sdk.schema.extract.payload.subject` | string | payload | no | The subject of the payload schema. If the record metadata contains the field "opencdc.collection" it is prepended to the subject name and separated with a dot. | |
| `sdk.schema.extract.type` | string | avro | no | The type of the payload schema. | inclusion(avro) |
| `servers` | string | | yes | Servers is a list of Kafka bootstrap servers, which will be used to discover all the servers in a cluster. | required |
| `tls.enabled` | bool | | no | TLSEnabled defines whether TLS is needed to communicate with the Kafka cluster. | |
| `topics` | string | | yes | Topics is a comma separated list of Kafka topics to read from. | required |
#### Destination parameters
| Key | Type | Default | Required | Description | Validations |
| --- | --- | --- | --- | --- | --- |
| `acks` | string | all | no | Acks defines the number of acknowledges from partition replicas required before receiving a response to a produce request. None = fire and forget, one = wait for the leader to acknowledge the writes, all = wait for the full ISR to acknowledge the writes. | inclusion(none,one,all) |
| `batchBytes` | int | 1000012 | no | BatchBytes limits the maximum size of a request in bytes before being sent to a partition. This mirrors Kafka's max.message.bytes. | |
| `caCert` | string | | no | CACert is the Kafka broker's certificate. | |
| `clientCert` | string | | no | ClientCert is the Kafka client's certificate. | |
| `clientID` | string | conduit-connector-kafka | no | ClientID is a unique identifier for client connections established by this connector. | |
| `clientKey` | string | | no | ClientKey is the Kafka client's private key. | |
| `compression` | string | snappy | no | Compression set the compression codec to be used to compress messages. | inclusion(none,gzip,snappy,lz4,zstd) |
| `deliveryTimeout` | duration | | no | DeliveryTimeout for write operation performed by the Writer. | |
| `insecureSkipVerify` | bool | | no | InsecureSkipVerify defines whether to validate the broker's certificate chain and host name. If 'true', accepts any certificate presented by the server and any host name in that certificate. | |
| `saslMechanism` | string | | no | Mechanism configures the connector to use SASL authentication. If empty, no authentication will be performed. | inclusion(PLAIN,SCRAM-SHA-256,SCRAM-SHA-512) |
| `saslPassword` | string | | no | Password sets up the password used with SASL authentication. | |
| `saslUsername` | string | | no | Username sets up the username used with SASL authentication. | |
| `sdk.batch.delay` | duration | 0 | no | Maximum delay before an incomplete batch is written to the destination. | |
| `sdk.batch.size` | int | 0 | no | Maximum size of batch before it gets written to the destination. | greater-than(-1) |
| `sdk.rate.burst` | int | 0 | no | Allow bursts of at most X records (0 or less means that bursts are not limited). Only takes effect if a rate limit per second is set. Note that if `sdk.batch.size` is bigger than `sdk.rate.burst`, the effective batch size will be equal to `sdk.rate.burst`. | greater-than(-1) |
| `sdk.rate.perSecond` | float | 0 | no | Maximum number of records written per second (0 means no rate limit). | greater-than(-1) |
| `sdk.record.format` | string | opencdc/json | no | The format of the output record. See the Conduit documentation for a full list of supported formats (https://conduit.io/docs/using/connectors/configuration-parameters/output-format). | |
| `sdk.record.format.options` | string | | no | Options to configure the chosen output record format. Options are normally key=value pairs separated with comma (e.g. opt1=val2,opt2=val2), except for the `template` record format, where options are a Go template. | |
| `sdk.schema.extract.key.enabled` | bool | true | no | Whether to extract and decode the record key with a schema. | |
| `sdk.schema.extract.payload.enabled` | bool | true | no | Whether to extract and decode the record payload with a schema. | |
| `servers` | string | | yes | Servers is a list of Kafka bootstrap servers, which will be used to discover all the servers in a cluster. | required |
| `tls.enabled` | bool | | no | TLSEnabled defines whether TLS is needed to communicate with the Kafka cluster. | |
| `topic` | string | {{ index .Metadata "opencdc.collection" }} | no | Topic is the Kafka topic. It can contain a [Go template](https://pkg.go.dev/text/template) that will be executed for each record to determine the topic. By default, the topic is the value of the `opencdc.collection` metadata field. | |
### log (v0.7.3)
A destination connector that logs all incoming records.
**Destination**
The destination connector logs records using the built-in Conduit logger.
Example log message:
```
{"level":"info","record":{"position":"cjI=","operation":"create","metadata":{"foo":"bar","opencdc.version":"v1"},"key":{"my-id-field":1},"payload":{"before":nil,"after":{"my-payload-field":false}}}}
```
Note that the `position` field is base64 encoded, same goes for a key or payload
that contains raw byte data.
Keep in mind that Conduit's log level needs to be configured lower or equal to
the log level of the connector in order for the records to show up in the logs.
**Known issues**
- **Only when using as a [standalone connector](https://conduit.io/docs/core-concepts#standalone-connector)**: Messages larger than 64KB will not be logged when using `log.level` as `info`.
This is a known issue caused by the default buffer value of `go-plugin`. More information can be found in [this comment](https://github.com/ConduitIO/conduit-connector-log/issues/81#issuecomment-2904224580).
Author: Meroxa, Inc.
#### Source parameters
(none)
#### Destination parameters
| Key | Type | Default | Required | Description | Validations |
| --- | --- | --- | --- | --- | --- |
| `level` | string | info | no | The log level used to log records. | inclusion(trace,debug,info,warn,error) |
| `message` | string | | no | Optional message that should be added to the log output of every record. | |
| `sdk.batch.delay` | duration | 0 | no | Maximum delay before an incomplete batch is written to the destination. | |
| `sdk.batch.size` | int | 0 | no | Maximum size of batch before it gets written to the destination. | greater-than(-1) |
| `sdk.rate.burst` | int | 0 | no | Allow bursts of at most X records (0 or less means that bursts are not limited). Only takes effect if a rate limit per second is set. Note that if `sdk.batch.size` is bigger than `sdk.rate.burst`, the effective batch size will be equal to `sdk.rate.burst`. | greater-than(-1) |
| `sdk.rate.perSecond` | float | 0 | no | Maximum number of records written per second (0 means no rate limit). | greater-than(-1) |
| `sdk.record.format` | string | opencdc/json | no | The format of the output record. See the Conduit documentation for a full list of supported formats (https://conduit.io/docs/using/connectors/configuration-parameters/output-format). | |
| `sdk.record.format.options` | string | | no | Options to configure the chosen output record format. Options are normally key=value pairs separated with comma (e.g. opt1=val2,opt2=val2), except for the `template` record format, where options are a Go template. | |
| `sdk.schema.extract.key.enabled` | bool | true | no | Whether to extract and decode the record key with a schema. | |
| `sdk.schema.extract.payload.enabled` | bool | true | no | Whether to extract and decode the record payload with a schema. | |
### postgres (v0.14.0)
Conduit connector for PostgreSQL
**Source**
The Postgres Source Connector connects to a database with the provided `url` and
starts creating records for each change detected in the provided tables.
Upon starting, the source takes a snapshot of the provided tables in the database,
then switches into CDC mode. In CDC mode, the plugin reads from a buffer of CDC events.
**Snapshot**
When the connector first starts, snapshot mode is enabled. The connector acquires
a read-only lock on the tables, and then reads all rows of the tables into Conduit.
Once all rows in that initial snapshot are read the connector releases its lock and
switches into CDC mode.
This behavior is enabled by default, but can be turned off by adding
`"snapshotMode": "never"` to the Source configuration.
**Change Data Capture**
This connector implements Change Data Capture (CDC) features for PostgreSQL by
creating a logical replication slot and a publication that listens to changes in the
configured tables. Every detected change is converted into a record. If there are no
records available, the connector blocks until a record is available or the connector
receives a stop signal.
**Logical Replication Configuration**
When the connector switches to CDC mode, it attempts to run the initial setup commands
to create its logical replication slot and publication. It will connect to an existing
slot if one with the configured name exists.
The Postgres user specified in the connection URL must have sufficient privileges to
run all of these setup commands, or it will fail.
Example pipeline configuration that's using logical replication:
```yaml
version: 2.2
pipelines:
- id: pg-to-log
status: running
connectors:
- id: pg
type: source
plugin: builtin:postgres
settings:
url: "postgres://exampleuser:examplepass@localhost:5433/exampledb?sslmode=disable"
tables: "users"
cdcMode: "logrepl"
logrepl.publicationName: "examplepub"
logrepl.slotName: "exampleslot"
- id: log
type: destination
plugin: builtin:log
settings:
level: info
```
:warning: When the connector or pipeline is deleted, the connector will automatically
attempt to delete the replication slot and publication. This is the default behaviour
and can be disabled by setting `logrepl.autoCleanup` to `false`.
**Key Handling**
The connector will automatically look up the primary key column for the specified tables
and use them as the key value. If that can't be determined, the connector will return
an error.
**Destination**
The Postgres Destination takes a Conduit record and stores it using a SQL statement.
The Destination is designed to handle different payloads and keys. Because of this,
each record is individually parsed and upserted.
**Handling record operations**
Based on the `Operation` field in the record, the destination will either insert,
update or delete the record in the target table. Snapshot records are always inserted.
If the target table already contains a record with the same key as a record being
inserted, the record will be updated (upserted). This can overwrite and thus potentially
lose data, so keys should be assigned correctly from the Source.
If the target table does not contain a record with the same key as a record being
deleted, the record will be ignored.
If there is no key, the record will be simply appended.
Author: Meroxa, Inc.
#### Source parameters
| Key | Type | Default | Required | Description | Validations |
| --- | --- | --- | --- | --- | --- |
| `cdcMode` | string | auto | no | CDCMode determines how the connector should listen to changes. | inclusion(auto,logrepl) |
| `logrepl.autoCleanup` | bool | true | no | LogreplAutoCleanup determines if the replication slot and publication should be removed when the connector is deleted. | |
| `logrepl.publicationName` | string | conduitpub | no | LogreplPublicationName determines the publication name in case the connector uses logical replication to listen to changes (see CDCMode). | |
| `logrepl.slotName` | string | conduitslot | no | LogreplSlotName determines the replication slot name in case the connector uses logical replication to listen to changes (see CDCMode). Can only contain lower-case letters, numbers, and the underscore character. | regex(^[a-z0-9_]+$) |
| `logrepl.withAvroSchema` | bool | true | no | WithAvroSchema determines whether the connector should attach an avro schema on each record. | |
| `sdk.batch.delay` | duration | 0 | no | Maximum delay before an incomplete batch is read from the source. | |
| `sdk.batch.size` | int | 0 | no | Maximum size of batch before it gets read from the source. | greater-than(-1) |
| `sdk.schema.context.enabled` | bool | true | no | Specifies whether to use a schema context name. If set to false, no schema context name will be used, and schemas will be saved with the subject name specified in the connector (not safe because of name conflicts). | |
| `sdk.schema.context.name` | string | | no | Schema context name to be used. Used as a prefix for all schema subject names. If empty, defaults to the connector ID. | |
| `sdk.schema.extract.key.enabled` | bool | false | no | Whether to extract and encode the record key with a schema. | |
| `sdk.schema.extract.key.subject` | string | key | no | The subject of the key schema. If the record metadata contains the field "opencdc.collection" it is prepended to the subject name and separated with a dot. | |
| `sdk.schema.extract.payload.enabled` | bool | false | no | Whether to extract and encode the record payload with a schema. | |
| `sdk.schema.extract.payload.subject` | string | payload | no | The subject of the payload schema. If the record metadata contains the field "opencdc.collection" it is prepended to the subject name and separated with a dot. | |
| `sdk.schema.extract.type` | string | avro | no | The type of the payload schema. | inclusion(avro) |
| `snapshot.fetchSize` | int | 50000 | no | Snapshot fetcher size determines the number of rows to retrieve at a time. | |
| `snapshotMode` | string | initial | no | SnapshotMode is whether the plugin will take a snapshot of the entire table before starting cdc mode. | inclusion(initial,never) |
| `tables` | string | | yes | Tables is a List of table names to read from, separated by a comma, e.g.:"table1,table2". Use "*" if you'd like to listen to all tables. | required |
| `url` | string | | yes | URL is the connection string for the Postgres database. | required |
#### Destination parameters
| Key | Type | Default | Required | Description | Validations |
| --- | --- | --- | --- | --- | --- |
| `sdk.batch.delay` | duration | 0 | no | Maximum delay before an incomplete batch is written to the destination. | |
| `sdk.batch.size` | int | 0 | no | Maximum size of batch before it gets written to the destination. | greater-than(-1) |
| `sdk.rate.burst` | int | 0 | no | Allow bursts of at most X records (0 or less means that bursts are not limited). Only takes effect if a rate limit per second is set. Note that if `sdk.batch.size` is bigger than `sdk.rate.burst`, the effective batch size will be equal to `sdk.rate.burst`. | greater-than(-1) |
| `sdk.rate.perSecond` | float | 0 | no | Maximum number of records written per second (0 means no rate limit). | greater-than(-1) |
| `sdk.record.format` | string | opencdc/json | no | The format of the output record. See the Conduit documentation for a full list of supported formats (https://conduit.io/docs/using/connectors/configuration-parameters/output-format). | |
| `sdk.record.format.options` | string | | no | Options to configure the chosen output record format. Options are normally key=value pairs separated with comma (e.g. opt1=val2,opt2=val2), except for the `template` record format, where options are a Go template. | |
| `sdk.schema.extract.key.enabled` | bool | true | no | Whether to extract and decode the record key with a schema. | |
| `sdk.schema.extract.payload.enabled` | bool | true | no | Whether to extract and decode the record payload with a schema. | |
| `table` | string | {{ index .Metadata "opencdc.collection" }} | no | Table is used as the target table into which records are inserted. | |
| `url` | string | | yes | URL is the connection string for the Postgres database. | required |
### s3 (v0.9.3)
An S3 source and destination plugin for Conduit, written in Go.
The S3 connector is one of [Conduit](https://github.com/ConduitIO/conduit)
builtin plugins. It provides both, a source and a destination S3 connectors.
**Source**
The S3 Source Connector connects to a S3 bucket with the provided
configurations, using `aws.bucket`, `aws.accessKeyId`,`aws.secretAccessKey` and
`aws.region`. If the bucket doesn't exist, or the permissions fail, then an
error will occur. After that, the `Open` method is called to start the
connection from the provided position.
**Change Data Capture (CDC)**
This connector implements CDC features for S3 by scanning the bucket for changes
every `pollingPeriod` and detecting any change that happened after a certain
timestamp. These changes (update, delete, create) are then inserted into a
buffer that is checked on each Read request.
* To capture "delete" and "update", the S3 bucket versioning must be enabled.
* To capture "create" actions, the bucket versioning doesn't matter.
**Position Handling**
The connector goes through two modes.
* Snapshot mode: which loops through the S3 bucket and returns the objects that
are already in there. The _Position_ during this mode is the object key
attached to an underscore, an "s" for snapshot, and the _maxLastModifiedDate_
found so far. As an example: "thisIsAKey_s12345", which makes the connector
know at what mode it is and what object it last read. The
_maxLastModifiedDate_ will be used when changing to CDC mode, the iterator
will capture changes that happened after that.
* CDC mode: this mode iterates through the S3 bucket every `pollingPeriod` and
captures new actions made on the bucket. the _Position_ during this mode is
the object key attached to an underscore, a "c" for CDC, and the object's
_lastModifiedDate_ in seconds. As an example: "thisIsAKey_c1634049397". This
position is used to return only the actions with a _lastModifiedDate_ higher
than the last record returned, which will ensure that no duplications are in
place.
**Record Keys**
The S3 object key uniquely identifies the objects in an Amazon S3 bucket, which
is why a record key is the key read from the S3 bucket.
**Destination**
The S3 Destination Connector connects to an S3 bucket with the provided
configurations, using `aws.bucket`, `aws.accessKeyId`,`aws.secretAccessKey` and
`aws.region`. If the permissions fail, the connector will not be ready for
writing to S3.
**Writer**
The S3 destination writer has a buffer with the size of `bufferSize`, for each
time `Write` is called, a new record is added to the buffer. When the buffer is
full, all the records from it will be written to the S3 bucket, and an ack
function will be called for each record after being written.
Author: Meroxa, Inc.
#### Source parameters
| Key | Type | Default | Required | Description | Validations |
| --- | --- | --- | --- | --- | --- |
| `aws.accessKeyId` | string | | yes | AWS access key id. | required |
| `aws.bucket` | string | | yes | the AWS S3 bucket name. | required |
| `aws.region` | string | | yes | the AWS S3 bucket region | required |
| `aws.secretAccessKey` | string | | yes | AWS secret access key. | required |
| `pollingPeriod` | duration | 1s | no | polling period for the CDC mode, formatted as a time.Duration string. | |
| `prefix` | string | | no | the S3 key prefix. | |
| `sdk.batch.delay` | duration | 0 | no | Maximum delay before an incomplete batch is read from the source. | |
| `sdk.batch.size` | int | 0 | no | Maximum size of batch before it gets read from the source. | greater-than(-1) |
| `sdk.schema.context.enabled` | bool | true | no | Specifies whether to use a schema context name. If set to false, no schema context name will be used, and schemas will be saved with the subject name specified in the connector (not safe because of name conflicts). | |
| `sdk.schema.context.name` | string | | no | Schema context name to be used. Used as a prefix for all schema subject names. If empty, defaults to the connector ID. | |
| `sdk.schema.extract.key.enabled` | bool | false | no | Whether to extract and encode the record key with a schema. | |
| `sdk.schema.extract.key.subject` | string | key | no | The subject of the key schema. If the record metadata contains the field "opencdc.collection" it is prepended to the subject name and separated with a dot. | |
| `sdk.schema.extract.payload.enabled` | bool | false | no | Whether to extract and encode the record payload with a schema. | |
| `sdk.schema.extract.payload.subject` | string | payload | no | The subject of the payload schema. If the record metadata contains the field "opencdc.collection" it is prepended to the subject name and separated with a dot. | |
| `sdk.schema.extract.type` | string | avro | no | The type of the payload schema. | inclusion(avro) |
#### Destination parameters
| Key | Type | Default | Required | Description | Validations |
| --- | --- | --- | --- | --- | --- |
| `aws.accessKeyId` | string | | yes | AWS access key id. | required |
| `aws.bucket` | string | | yes | the AWS S3 bucket name. | required |
| `aws.region` | string | | yes | the AWS S3 bucket region | required |
| `aws.secretAccessKey` | string | | yes | AWS secret access key. | required |
| `format` | string | | yes | the destination format, either "json" or "parquet". | required, inclusion(parquet,json) |
| `prefix` | string | | no | the S3 key prefix. | |
| `sdk.batch.delay` | duration | 0 | no | Maximum delay before an incomplete batch is written to the destination. | |
| `sdk.batch.size` | int | 0 | no | Maximum size of batch before it gets written to the destination. | greater-than(-1) |
| `sdk.rate.burst` | int | 0 | no | Allow bursts of at most X records (0 or less means that bursts are not limited). Only takes effect if a rate limit per second is set. Note that if `sdk.batch.size` is bigger than `sdk.rate.burst`, the effective batch size will be equal to `sdk.rate.burst`. | greater-than(-1) |
| `sdk.rate.perSecond` | float | 0 | no | Maximum number of records written per second (0 means no rate limit). | greater-than(-1) |
| `sdk.record.format` | string | opencdc/json | no | The format of the output record. See the Conduit documentation for a full list of supported formats (https://conduit.io/docs/using/connectors/configuration-parameters/output-format). | |
| `sdk.record.format.options` | string | | no | Options to configure the chosen output record format. Options are normally key=value pairs separated with comma (e.g. opt1=val2,opt2=val2), except for the `template` record format, where options are a Go template. | |
| `sdk.schema.extract.key.enabled` | bool | true | no | Whether to extract and decode the record key with a schema. | |
| `sdk.schema.extract.payload.enabled` | bool | true | no | Whether to extract and decode the record payload with a schema. | |
## 3. Error codes (81)
| Reason | gRPC status | Description |
| --- | --- | --- |
| `common.invalid_argument` | InvalidArgument | CodeInvalidArgument is a generic invalid-input error. |
| `common.not_found` | NotFound | CodeNotFound is a generic not-found error. |
| `common.unavailable` | Unavailable | common: unavailable |
| `config.field_invalid` | InvalidArgument | config: field invalid |
| `config.field_renamed` | InvalidArgument | CodeFieldRenamed is the stable code carried by a lint warning for a deprecated field that was mechanically renamed (the old key's value moves unchanged to a new key) — see pkg/provisioning/config/yaml/internal.Change.RenamedTo and (*configLinter).newWarning. It is distinct from the generic validate.CodeLintWarning so cmd/conduit/internal/repair's fix-producer scan (design doc 20260712-repair-command.md §6, item #1) can find exactly these findings without pattern-matching warning text. |
| `config.field_required` | InvalidArgument | config: field required |
| `config.field_too_long` | InvalidArgument | config: field too long |
| `config.id_duplicate` | InvalidArgument | config: id duplicate |
| `config.parse_error` | InvalidArgument | CodeParseError is raised when a pipeline config document can't be parsed at all (invalid YAML, unrecognized version). It exists for offline consumers (cmd/conduit/internal/validate) that need a stable code for parser failures, which the parser itself returns as plain errors since it has no per-field configPath to attach at that stage. |
| `connector.instance_not_found` | NotFound | CodeConnectorNotFound is raised when a referenced connector instance cannot be located. |
| `connector.invalid_type` | InvalidArgument | CodeConnectorInvalidType is raised when a connector type is neither "source" nor "destination". |
| `connector.plugin_not_found` | NotFound | CodeConnectorPluginNotFound is raised when a referenced connector plugin cannot be located. |
| `connector.running` | FailedPrecondition | CodeConnectorRunning is raised when an operation requires the connector to not be running, but a connector instance for it already exists. |
| `internal.error` | Internal | CodeInternal is a coded internal failure. |
| `internal.unknown` | Internal | CodeUnknown is the fallback when an un-coded error reaches a boundary. |
| `lifecycle_v2.stop_and_wait_unsupported` | Unimplemented | lifecycle v2: stop and wait unsupported |
| `orchestrator.connector_has_processors_attached` | FailedPrecondition | CodeConnectorHasProcessorsAttached is raised when a connector cannot be deleted because it still has processors attached. |
| `orchestrator.immutable_provisioned_by_config` | FailedPrecondition | CodeImmutableProvisionedByConfig is raised when a pipeline, connector, or processor provisioned by a config file is mutated through the API instead of the config file. |
| `orchestrator.invalid_processor_parent_type` | InvalidArgument | CodeInvalidProcessorParentType is raised when a processor's parent is neither a pipeline nor a connector. |
| `orchestrator.pipeline_has_connectors_attached` | FailedPrecondition | CodePipelineHasConnectorsAttached is raised when a pipeline cannot be deleted because it still has connectors attached. |
| `orchestrator.pipeline_has_processors_attached` | FailedPrecondition | CodePipelineHasProcessorsAttached is raised when a pipeline cannot be deleted because it still has processors attached. |
| `pipeline.instance_not_found` | NotFound | CodePipelineNotFound is raised when a referenced pipeline instance cannot be located. |
| `pipeline.name_already_exists` | AlreadyExists | CodePipelineNameAlreadyExists is raised when a pipeline config's name collides with an existing pipeline's name. |
| `pipeline.name_missing` | InvalidArgument | CodePipelineNameMissing is raised when a pipeline config is missing the required name field. |
| `pipeline.not_running` | FailedPrecondition | CodePipelineNotRunning is raised when an operation requires the pipeline to be running, but it is currently stopped. |
| `pipeline.running` | FailedPrecondition | CodePipelineRunning is raised when an operation requires the pipeline to be stopped, but it is currently running. |
| `pipelines.init_destination_exists` | AlreadyExists | pipelines: init destination exists |
| `pipelines.init_template_flags_exclusive` | InvalidArgument | pipelines: init template flags exclusive |
| `pipelines.init_template_version_mismatch` | FailedPrecondition | pipelines: init template version mismatch |
| `pipelines.init_unknown_template` | InvalidArgument | pipelines: init unknown template |
| `processor.instance_not_found` | NotFound | CodeProcessorNotFound is raised when a referenced processor instance cannot be located. |
| `processor.plugin_not_found` | NotFound | CodeProcessorPluginNotFound is raised when a referenced processor plugin cannot be located. |
| `provisioning.live_apply_unauthorized` | FailedPrecondition | provisioning: live apply unauthorized |
| `provisioning.multi_pipeline_file` | InvalidArgument | provisioning: multi pipeline file |
| `provisioning.pipeline_id_duplicate` | InvalidArgument | provisioning: pipeline id duplicate |
| `provisioning.pipeline_running` | FailedPrecondition | provisioning: pipeline running |
| `provisioning.plan_stale` | FailedPrecondition | provisioning: plan stale |
| `provisioning.standalone_unsafe` | FailedPrecondition | provisioning: standalone unsafe |
| `registry.ambiguous_uninstall` | InvalidArgument | CodeAmbiguousUninstall is raised when "uninstall <name>" is given with no version and more than one version of that name is installed (§3.1 — a new behavior versus every original step-plan, introduced by the manifest's name@version key change). |
| `registry.archive_invalid` | Internal | CodeArchiveInvalid is raised when a downloaded archive has zero or more than one candidate executable, a path-traversal entry, or a symlink entry. |
| `registry.bundle_stale` | FailedPrecondition | CodeBundleStale is raised when an offline --bundle is older than maxStaleness and --allow-stale-bundle was not set. |
| `registry.bundle_too_large` | ResourceExhausted | CodeBundleTooLarge is raised when a signature or provenance bundle fetch exceeds the P0-2 size cap. ResourceExhausted already carries a defined pkg/conduit/exitcode bucket (Environment) — see index.CodeIndexTooLarge's doc comment for the same confirmation. |
| `registry.connector_in_use` | FailedPrecondition | CodeConnectorInUse is raised when an uninstall is refused because a pipeline still references the exact name@version. |
| `registry.connector_not_found` | NotFound | CodeConnectorNotFound is raised when an exact-match name lookup in the index fails. |
| `registry.connector_not_installed` | NotFound | CodeConnectorNotInstalled is raised on an uninstall/manifest lookup miss. |
| `registry.corrupt_download` | DataLoss | CodeCorruptDownload is raised when the received-bytes sha256 does not match the index's declared sha256 — DataLoss, matching gRPC's literal "unrecoverable data loss or corruption" semantics for a digest mismatch; a retry is a fresh download attempt, not a recovery of the same lost bytes. |
| `registry.download_failed` | Unavailable | CodeDownloadFailed is raised on a non-2xx artifact download, a redirect loop, or a connection reset. |
| `registry.identity_mismatch` | PermissionDenied | CodeIdentityMismatch is raised when a signature verifies but for a different certificate identity than the one pinned (trust.ErrIdentityMismatch). |
| `registry.identity_pattern_too_loose` | PermissionDenied | CodeIdentityPatternTooLoose is raised when a connector's pinned Publisher.ExpectedIdentityPattern fails ValidateIdentityPattern's tightness rules at verify time (trust.ErrIdentityPatternTooLoose) — defense-in-depth against a signature-verified index whose pinned pattern is dangerously loose (e.g. unanchored, or "^.*$"), which would otherwise let an arbitrary signing identity satisfy the pin. |
| `registry.identity_revoked` | PermissionDenied | CodeIdentityRevoked is raised when a connector's publisher carries `revoked` — every version under that name is refused regardless of individual yanked status. audit's REVOKED_PUBLISHER finding (PR-4) reuses this code verbatim. |
| `registry.incompatible_version` | FailedPrecondition | CodeIncompatibleVersion is raised when an explicit @version pin is below the running Conduit's min-versions, or (see ManifestKey) when a version string fails to parse as semver at all. |
| `registry.index_integrity` | DataLoss | CodeIndexIntegrity is raised when a recognized keyId's cryptographic verification fails (tampering/corruption) — and, per this package's duplicate-key walker, also when parse-time duplicate-key rejection fires: R-1 frames duplicate-key resolution ambiguity as itself a "signature-bypass primitive" (a producer/verifier parser differential), which is the same integrity concern this code names, not a separate condition. See duplicatekey.go's doc comment. |
| `registry.index_nesting_too_deep` | FailedPrecondition | CodeIndexNestingTooDeep is raised when the duplicate-key walker's recursion cap is hit (P0-2 item 2) — refuse, never stack-overflow. |
| `registry.index_rollback` | FailedPrecondition | CodeIndexRollback is raised when index.version is below the locally persisted high-water mark. |
| `registry.index_stale` | FailedPrecondition | CodeIndexStale is raised when index.timestamp is older than maxStaleness — distinct from CodeIndexUnreachable and CodeIndexRollback. |
| `registry.index_too_large` | ResourceExhausted | CodeIndexTooLarge is raised when a fetched index exceeds the P0-2 size cap (plan-v2 §2.4 item 1). ResourceExhausted already carries a defined pkg/conduit/exitcode bucket (Environment) — see fetch.go's doc comment for the footnote this resolves. |
| `registry.index_unreachable` | Unavailable | CodeIndexUnreachable is raised when fetching the index fails at the network/HTTP layer (distinct from a fetch that succeeds but is too large, stale, or rolled back). |
| `registry.install_locked` | Unavailable | CodeInstallLocked is raised when the per-target install lock is not acquired within --lock-timeout. |
| `registry.no_platform_artifact` | NotFound | CodeNoPlatformArtifact is raised when no artifact exists for the host (os, arch). |
| `registry.provenance_invalid` | PermissionDenied | CodeProvenanceInvalid is raised when the SLSA provenance's subject-digest match or builder.id/configSource.uri binding fails (trust.ErrProvenanceInvalid). |
| `registry.schema_too_new` | FailedPrecondition | CodeSchemaTooNew is raised when payload.schemaVersion exceeds the highest this build was compiled to understand. |
| `registry.trust_anchor_expired` | FailedPrecondition | CodeTrustAnchorExpired is raised when no keyId in signatures[] matches any of this build's compiled-in trust anchors at all — the "upgrade Conduit" case, distinct from CodeIndexIntegrity's "recognized key, verification failed". |
| `registry.trust_anchors_unavailable` | Internal | CodeTrustAnchorsUnavailable is raised when this conduit build's compiled-in registry trust anchors could not be loaded (the embedded public keys are missing or unparseable) — a build/release defect, not a normal expired-anchor condition. It is codes.Internal (not the user's input to fix) and machine-signals "reinstall a release build of conduit," never "retry". Distinct from index.CodeTrustAnchorExpired, which means the build HAS anchors but the fetched index was signed by a key not among them. Verification never proceeds when this is raised — it can only make installs fail closed, never open. |
| `registry.unsigned` | PermissionDenied | CodeUnsigned is raised when there is no valid signature for the pinned identity at all (trust.ErrUnsigned). |
| `registry.unsigned_install_disabled_by_policy` | PermissionDenied | CodeUnsignedInstallDisabledByPolicy is raised when operator policy (install.allowUnsigned: false) hard-disables the gate regardless of flag/TTY/env var. |
| `registry.unsigned_install_non_interactive` | PermissionDenied | CodeUnsignedInstallNonInteractive is raised when --allow-unsigned was requested in a non-interactive context (no TTY, CI=true, or MCP-originated) without the escape-hatch env var also set, or when an interactive typed confirmation was not given. |
| `registry.verification_unavailable` | Unimplemented | CodeVerificationUnavailable is raised by FailClosedVerifier — no longer this package's production wiring as of PR-2 (see TrustedVerifier), but still the correct, intentional behavior for any caller that explicitly constructs Install with FailClosedVerifier (e.g. a build with the trust core deliberately disabled), and for the structural belt-and-suspenders refusal if a future ArtifactVerifier implementation ever returns success without actually signing. |
| `registry.version_not_found` | NotFound | CodeVersionNotFound is raised when the connector exists but the requested @version does not. |
| `registry.version_yanked` | FailedPrecondition | CodeVersionYanked is raised when a resolved/pinned version carries `yanked`. Registered here (owning package "index" per plan-v2 §4, which lists it as shared with "registry") rather than in pkg/registry, specifically so this package need not import pkg/registry (which imports this package) — audit's REVOKED_PUBLISHER finding (PR-4) reuses this code verbatim rather than minting a new connector.*-prefixed one. |
| `repair.ambiguous_fix` | FailedPrecondition | CodeAmbiguousFix is raised when more than one candidate fix targets the same ConfigPath and the selection does not disambiguate — repair never silently picks one (design doc §9, failure mode 3, AC-20). |
| `repair.data_path_fix_refused` | FailedPrecondition | CodeDataPathFixRefused is raised when a FixClassDataPath fix is selected for apply without the human-only Escalate path (design doc §4.2, the Tier-1 crux). The CLI surfaces this as a hard refusal (AC-14); MCP repair_apply — which never sets Escalate — surfaces it as a per-fix skip, never a fatal call (AC-15). |
| `repair.fix_no_longer_applies` | FailedPrecondition | CodeFixNoLongerApplies is raised per-fix (not for the whole Apply call) when a selected fix's target field no longer matches what the fix was computed against — either hand-edited between Collect and Apply in a way the hash's per-fix granularity didn't already catch, or consumed by an earlier fix in the same Apply batch (design doc §9, failure mode 1 and AC-19). Other selected fixes in the same call still apply; this fix is skipped, not fatal to the run. |
| `repair.no_fixes_available` | InvalidArgument | CodeNoFixesAvailable is raised when --apply (or repair_apply) is requested but the plan contains zero appliable fixes. A read-mode call with zero fixes is exit 0 ("already clean"), never this code (AC-21). |
| `repair.plan_stale` | FailedPrecondition | CodePlanStale is raised by Apply when the caller-presented hash does not match the hash freshly recomputed from the current file bytes — the direct analogue of provisioning.CodePlanStale (see plan.go). The file changed since the plan was shown and approved. |
| `scaffold.build_failed` | Internal | CodeBuildFailed is raised when the verified `go build ./...` step fails. Maps to gRPC Internal -> exitcode.Runtime (1), per this package's explicit exit-code routing ("verified go build failure -> 1"). |
| `scaffold.destination_exists` | AlreadyExists | CodeDestinationExists is raised when the target directory already exists and --force was not given. Maps to gRPC AlreadyExists -> exitcode.Validation (2). |
| `scaffold.generate_failed` | Internal | CodeGenerateFailed is raised when installing the code-gen tool or running it fails after files are already on disk (in the temp staging directory — see scaffold.go's atomic-rename discipline, so this never leaves a partial directory at the requested destination). Maps to gRPC Internal -> exitcode.Runtime (1). |
| `scaffold.invalid_module` | InvalidArgument | CodeInvalidModule is raised when --module is missing (and cannot be prompted for) or does not match the "github.com/<owner>/conduit-(connector\|processor)-<name>" shape a template's own setup.sh requires. Maps to gRPC InvalidArgument -> exitcode.Validation (2). |
| `scaffold.toolchain_unavailable` | Unavailable | CodeToolchainUnavailable is raised when the preflight (Go on PATH at the minimum version, git on PATH, GOPATH/bin writable) fails. Maps to gRPC Unavailable -> exitcode.Environment (3): the fix is "install or upgrade a toolchain component", not "change your command line". |
| `scaffold.unsupported_language` | InvalidArgument | CodeUnsupportedLanguage is raised for --lang values other than "go" — today that's every value, since Go is the only ready target (see the design doc's feasibility verdict; Python is blocked on a connector SDK that does not exist yet). Maps to gRPC InvalidArgument -> exitcode.Validation (2). |
| `scaffold.write_failed` | Internal | CodeWriteFailed is raised for internal I/O failures unpacking the embedded template snapshot into the staging directory, or performing the atomic rename into place. These are not user-fixable the way a bad flag or a missing toolchain is; they indicate something is wrong with the local filesystem itself. Maps to gRPC Internal -> exitcode.Runtime (1). |
## 4. MCP tool catalog (13)
| Name | Class | Description |
| --- | --- | --- |
| `apply` | write (requires --allow-mutations) | Applies the plan computed by deploy, only if hash still matches the freshly recomputed plan (a stale hash is refused, nothing mutated). Refuses to mutate a currently-running pipeline. Same engine as `conduit pipelines apply`. |
| `deploy` | read | Computes the diff (create/update/delete per pipeline/connector/processor) needed to deploy a pipeline config against its current stored state, and a plan hash to bind a later apply call to. Mutates nothing. Same engine as `conduit pipelines deploy`. |
| `doctor` | read | Checks whether the local machine and configuration are ready for `conduit run` — offline, non-destructive. Same engine as `conduit doctor`. |
| `dry_run` | read | Everything validate checks, plus the fully-enriched pipeline graph `conduit run` would load and a builtin-plugin existence check. Same engine as `conduit pipelines dry-run`. |
| `inspect` | read | Reports a running pipeline's live status, per-stage connector summary, and DLQ. Requires a running Conduit (conduit mcp --api-address). Same engine as `conduit pipelines inspect`. |
| `lint` | read | Everything validate checks, plus the parser's advisory warnings (deprecated/renamed/unknown fields, version fallback). Same engine as `conduit pipelines lint`. |
| `repair` | read | Scans a pipeline configuration for findings that carry a structured, machine-appliable fix (a deprecated/renamed field, an unambiguous invalid status value, a negative processor workers count, an over-long description), classifies each fix's safety (safe / restart / data_path), and returns a plan hash. Mutates nothing. Same engine as `conduit pipelines repair`. |
| `repair_apply` | write (requires --allow-mutations) | Applies the plan computed by repair — safe fixes only, unless select names others — only if hash still matches the freshly recomputed plan (a stale hash is refused, nothing mutated). A data-path-adjacent fix (connector settings, a connector's own plugin/type, DLQ config, any id field) is NEVER applied by this tool, even if selected explicitly — it comes back in the result as a skipped fix; only the CLI's --escalate flag (human-only) can apply one. Same engine as `conduit pipelines repair --apply`. |
| `scaffold_connector` | write (requires --allow-mutations) | Scaffolds a new Go connector plugin repository (source and/or destination). Same engine as `conduit connector new`. |
| `scaffold_processor` | write (requires --allow-mutations) | Scaffolds a new Go (WASM) processor plugin repository. Same engine as `conduit processor new`. |
| `start` | write (requires --allow-mutations) | Starts a pipeline registered in a running Conduit (transitions to Running). Requires --api-address, like inspect; no offline fallback. Refused if the pipeline is already running (pipeline.running). Same engine as `conduit pipelines start`. |
| `stop` | write (requires --allow-mutations) | Stops a running pipeline registered in a running Conduit — graceful drain by default, or immediate with force:true. Requires --api-address, like inspect; no offline fallback. Refused if the pipeline isn't running (pipeline.not_running). Same engine as `conduit pipelines stop`. |
| `validate` | read | Offline, static validation of a pipeline configuration (errors only, no side effects). Same engine as `conduit pipelines validate`. |