chore: add connector-review skill

[slbotbm]

Adds a connector-review skill that uses team-review's structure with the concerns outlined in the .claude/connector-* skills.

[slbotbm]

I don't have claude, so could someone please test this on one of the connectors PRs?

[mattp5657]

I can test it on my OpenSearch one later today.

[slbotbm]

@mattp5657 thanks!

[mattp5657]

Testing the connector-review skill on PR #3873 (OpenSearch sink)

Ran the skill end to end to validate it:

• 4 independent expert agents reviewed in parallel (plugin/sink, runtime/FFI, SDK contract, testing/docs), each reading the full diff and changed files with zero visibility into each other's findings.
• Experts produced 5 claims total, all nit/simplify severity, zero critical/warning.
• A clean-room validator, with no knowledge of which expert raised what, re-traced every claim against the source independently. All 5 passed on substance; 2 had corrections applied (one wrong line citation fixed, one origin tag fixed).
• A separate adversarial sweep agent, given only the final claim list and diff, hunted for anything missed or wrongly cleared. It added 5 more nit/simplify findings and confirmed none of the original claims wrongly dismissed a real bug.
• Final synthesis: 7 nits, 3 simplification suggestions, 0 blockers. Verdict: Approve.

Demonstrates the skill's core value: independent multi-angle review, adversarial self-checking that catches its own citation errors, and a sweep pass that guards against groupthink, all before a human reads a line. Full review below:

[mattp5657]

Connector Review - opensearch_sink (f3d9688)

Target: 3873 (apache/iggy, open, unmerged)
Reviewed commit: f3d9688d072e6b3337c7233a3fd2c5c353138abb
Timestamp: 2026-09-10T00:00:00Z (America/New_York)
Roles: plugin, runtime, sdk, testing
Validators: 1 shard validator (all 5 claims) + 1 sweep validator
Contested: 0

Review: add OpenSearch sink connector

Confirmed (expert + clean-room validator)

• [nit] opensearch_sink::index_documents (core/connectors/sinks/opensearch_sink/src/lib.rs:494) - multi-chunk consume() keeps the first chunk error, not the last, unlike http_sink/mongodb_sink/surrealdb_sink's last-err convention. Fix: none required, consistency-awareness only; indexed/failed counts are correct regardless of which error message is kept. (raised: plugin; validated: PASS)
• [nit] OpenSearchSink::retry_on_open (core/connectors/sinks/opensearch_sink/src/lib.rs:276) - a non-transient open()-time failure is returned raw, not wrapped in Error::InitError, unlike postgres_sink/mongodb_sink/http_sink/doris_sink/redshift_sink/surrealdb_sink/clickhouse_sink convention. Fix: wrap in Error::InitError at lib.rs:276 for consistency; behaviorally inert - the FFI boundary (sdk/src/sink.rs:89, runtime/src/sink.rs:459-475) collapses any Result to a generic i32/message regardless of variant, so even the error text is lost either way. (raised: plugin; validated: FIX - anchor corrected from the claimed :280/:338 (already-correct code) to the actual unwrapped-return site :276)
• [nit] OpenSearchSinkConfig (core/connectors/sinks/opensearch_sink/src/lib.rs:74-75) - comment falsely claims no in-tree redacted SecretString serializer exists; iggy_common::serde_secret::serialize_optional_redacted (core/common/src/utils/serde_secret.rs:85-93) predates this PR and does exactly that. Fix: reword the comment to state the real reason (simplicity of omitting Serialize entirely) rather than a false tooling claim; security posture is unaffected either way. (raised: sdk; validated: PASS)
• [nit] core/integration/tests/connectors/opensearch/{opensearch_sink.rs:33, opensearch_sink_failures.rs:132} - new integration tests use unit-test given_when_should naming instead of the connector-testing skill's mandated declarative <subject>_<action>_<observation> style. Fix: rename to match postgres_sink.rs::json_messages_sink_stores_as_bytea / elasticsearch_sink.rs::elasticsearch_sink_stores_json_messages. (raised: testing; validated: FIX - origin corrected pre-surfaced -> intro since both files are brand-new (new file mode 100644); framing corrected from "repo-wide drift" to "widens the one prior instance" (meilisearch_sink), since the declarative style is actually the dominant repo convention)
• [nit] OpenSearchSink::sleep_before_retry / index_documents (core/connectors/sinks/opensearch_sink/src/lib.rs:669-672, also :318,333,372,385,782,787) - retry/lifecycle/index-count logs omit connector ID and index name, unattributable when several plugin instances share one LOG_CALLBACK (exactly the setup the PR's own 3-instance failure test uses). Sibling arm at lib.rs:798-801 does carry self.id; the success-count arm at :782 drops it. Fix: add connector ID: {self.id} to every log site where &self is in scope. (sweep, unvalidated)
• [nit] OpenSearchSinkConfig (core/connectors/sinks/opensearch_sink/src/lib.rs:76) - derived Debug prints url verbatim; embedded credentials in url are rejected only later, in normalize_url, which never runs before From<OpenSearchSinkConfig>. ResolvedOpenSearchSinkConfig hand-redacts this field but the raw config does not, and the existing redaction test covers only password, not url. Latent - no in-tree caller Debug-formats the raw config today. Fix: hand-write Debug for OpenSearchSinkConfig through the existing redact_url_credentials helper, or drop the derive. (sweep, unvalidated)
• [nit] opensearch_sink::index_chunk (core/connectors/sinks/opensearch_sink/src/lib.rs:588) - the ? on build_bulk_body sits after the partial-merge step, so a failure there would discard the indexed count already credited from earlier attempts, causing already-indexed documents to be reported as failed. Unreachable today (serde_json::to_writer over this Value cannot fail), so latent rather than live. Fix: merge outcome into a returned Ok before propagating, or make body construction infallible. (sweep, unvalidated, conf:M)

Contested

None.

Retracted (validator REMOVE)

None.

Pre-existing (origin pre-*, not blocking)

None as findings against this PR. Two repo-wide gaps were independently checked by the runtime and plugin experts and confirmed correctly disclosed in the PR body rather than hidden: runtime/src/sink.rs:740 discards the FFI consume return value (so a plugin-side failure never reaches connector status//stats), and runtime/src/sink.rs:522 commits offsets via AutoCommit::When(PollingMessages) before consume runs. Both are pre-untouched by this diff and affect every sink identically - not something this PR introduces or is responsible for fixing.

Simplification opportunities (non-blocking)

• [simplify] sanitize_url_for_log (core/connectors/sinks/opensearch_sink/src/lib.rs:1296-1308) - credential-stripping branches are dead in practice: the sole call site always passes an already-sanitized normalized_url. Simpler: reduce to a to_string() alias. Saves: ~6 lines - but the code already self-documents this exact tradeoff as intentional defense-in-depth (lib.rs:1293-1295), and removing it orphans the unit test at :2861, so marginal value is low. (raised: plugin; validated: PASS, low marginal value)
• [simplify] opensearch_sink::documents_at (core/connectors/sinks/opensearch_sink/src/lib.rs:928-936) - out-of-range filter is dead: parse_bulk_response already rejects any response whose items.len() != expected before documents_at runs. Same dead-defense class as the item above, different function. Fix: none required; if kept, correct the doc comment's panic claim to reflect that the state it guards against is already excluded upstream. (sweep, unvalidated)
• [simplify] cluster_health / index_chunk (core/connectors/sinks/opensearch_sink/src/lib.rs:298, lib.rs:545) - per-call tokio::time::timeout duplicates the transport-level timeout already installed (lib.rs:248-249), which covers strictly more (including body reads); index_exists/create_index rely on the transport timeout alone, so the codebase is internally inconsistent about which layer owns the bound. Fix: drop the per-call wrappers, or document why two layers are intentional. (sweep, unvalidated, conf:M)

Verdict: APPROVE

Four independent experts (plugin, runtime, sdk, testing) plus clean-room validation and an adversarial sweep found zero critical or warning findings. Every confirmed item is a nit-level naming/logging/comment-accuracy issue or a latent (not live) gap; two are already self-documented by the author as intentional tradeoffs. runtime/src/ has no changes, no STOP-tripwire was triggered, and the mandatory connector checklist (SecretString, no Serialize, FFI codes, idempotency, retry/transient classification, drop accounting, forward-compat config, &self receivers, no blocking calls) verified clean across all four passes.

Counts: critical 0, warning 0, nit 7, simplify 3

---

Raw findings per expert

plugin

# Plugin review: PR 3873 (opensearch_sink) — role: plugin

Exemplars compared: `meilisearch_sink` (closest — same batch_size=1000 default,
same `Payload` dispatch shape, same `InvalidRecordValue` on unsupported
payload), `elasticsearch_sink` (closest sibling search-engine sink, same
`owned_value_to_serde_json` usage, same `close()` shape).

## Findings

[nit] opensearch_sink::index_documents (core/connectors/sinks/opensearch_sink/src/lib.rs:494) - keeps the *first* chunk error across a multi-chunk consume(), not the last. Every other multi-batch sink (`http_sink:595`, `mongodb_sink:241`, `surrealdb_sink:596`) uses a last-err accumulator. Deliberate and commented ("kept, not the last, matching BulkOutcome::merge... the one an operator can act on"), and every chunk is still processed and counted regardless of which error surfaces — counts (`indexed`/`failed`) are correct either way, only the reported message text differs. Fix: none required; flagging for consistency awareness only. (intro, conf:H)

[nit] OpenSearchSink::retry_on_open (core/connectors/sinks/opensearch_sink/src/lib.rs:263-290, call sites :280,:338) - a non-transient open()-time failure returns the raw `PermanentHttpError`/`Connection` error directly instead of wrapping it in `Error::InitError`, unlike every other sink's connectivity-failure convention (`postgres_sink:171`, `mongodb_sink:179-214`, `http_sink:347`, `doris_sink:253`, `redshift_sink:146`, `surrealdb_sink:362`, `clickhouse_sink:176` all wrap in `InitError`). Confirmed intentional and unit-tested (`given_permanent_open_failure_should_not_retry`, lib.rs:2259-2274, asserts `PermanentHttpError` survives untouched). Behaviorally inert: `manager/sink.rs` flips `ConnectorStatus::Error` on any `Err` from `open()` regardless of variant (sink.rs:106), so this is pure error-message classification, not a functional gap. (intro, conf:M)

[simplify] sanitize_url_for_log (core/connectors/sinks/opensearch_sink/src/lib.rs:1296-1308) - the username/password-stripping branches are dead in practice: its only call site (`open()`, lib.rs:685) always passes `normalized_url`, and `normalize_url` (lib.rs:1221) already rejects any URL with embedded credentials before it can reach here. Already documented as intentional defense-in-depth via the comment above the fn. Simpler: could be removed or reduced to a `to_string()` alias since the branch never fires today, but doing so would remove the safety net for any future caller that bypasses `normalize_url`. Not worth changing. Saves: ~6 lines, but conf:L this is worth doing. (intro, conf:L)

## Verified correct (no finding, checked because charter mandates it)

- `SecretString` on `password`; no `Serialize` on `OpenSearchSinkConfig` (comment at lib.rs:74 explains why — matches convention, e.g. no sink derives `Serialize` on its config type).
- Custom `Debug` impls on both `OpenSearchSink` and `ResolvedOpenSearchSinkConfig` redact the password and URL credentials (lib.rs:109-165) — needed because `opensearch::auth::Credentials::Basic` derives an unredacted `Debug`; verified via `given_debug_formatted_*_should_redact_password` tests (lib.rs:2988-3040).
- `consume(&self, ...)` — no interior `&mut self`, no locks, no `tokio::spawn`/`block_on`.
- `BTreeMap<HeaderKey, HeaderValue>` headers handled via `headers_to_json` (lib.rs:1113), and unlike the pre-existing gap the PR documents in `elasticsearch_sink`/`meilisearch_sink` (headers silently dropped because `serde_json` can't serialize a `HeaderKey`-keyed map), this sink actually converts and indexes them — a real fix, not a regression.
- `try_to_bytes`/no-clone JSON path: `prepare_document` uses `mem::replace` to take `message.payload` by value (lib.rs:408), then `owned_value_to_serde_json(&value)` (a reference-taking conversion helper already used identically by `elasticsearch_sink:195`) — matches SDK convention, not a clone-then-replace anti-pattern.
- `Vec::with_capacity(messages_count)` at lib.rs:741.
- `[lib] crate-type = ["cdylib", "lib"]` present in Cargo.toml.
- `verbose_logging` mirrors `debug!`→`info!` correctly in both `consume()` receive-log and indexed-count log (lib.rs:709-733, 780-790).
- 3-attempt cap (`DEFAULT_MAX_RETRIES = 3`), exponential backoff + jitter shared via SDK `retry` module, matches sibling sinks.
- Idempotency: dedup-on-write via deterministic `_id` (hash of stream/topic/partition/offset/message-id, or user `document_id_field`), verified end-to-end by `connectors::opensearch::opensearch_sink` (resends under existing `order_id`, asserts count unchanged) — correctly distinguished from the ES-auto-`_id` non-idempotent case the PR body calls out.
- No `unwrap()`/`expect()` on external-I/O `Result`s in production code (verified via grep over lib.rs:1-1350; all `unwrap()`/`expect()` usage is in `#[cfg(test)] mod tests`).
- Config forward-compat: every `OpenSearchSinkConfig` field is `Option<T>` with defaults resolved in `From<OpenSearchSinkConfig> for ResolvedOpenSearchSinkConfig` (lib.rs:167-203), conflict (retry_delay > max_retry_delay) fixed with `warn!` + swap in `new()`-time conversion, never in `consume()`.
- `Permanent*` vs transient classification: `is_transient_error`/`is_transient_client_error`/`map_status_error`/`map_client_error` all correctly route through `Error::HttpRequestFailed` (transient) vs `Error::PermanentHttpError` (permanent), including the per-item `_bulk` response classification (429/5xx retryable, everything else including mapping errors permanent) — unit-tested extensively (lib.rs:1971-2194).
- Drop accounting: `invalid_records`/`preparation_errors` both roll into `errors_count` (lib.rs:765-770); a single bad message in a batch does not lose the rest (comment at lib.rs:754 states this explicitly and `given_permanently_failing_chunk_should_not_abandon_later_chunks` proves the chunk-boundary case at the unit level, integration test `opensearch_sink_failures.rs` proves it against a live server).
- FFI/runtime consume()-swallow gap and `elasticsearch_sink`/`meilisearch_sink` header-drop gap: both correctly classified in the PR body as `pre-untouched` (verified against current `master` citations, e.g. `sink.rs:740-748`, `elasticsearch_sink/src/lib.rs:205-219`) — out of scope for this PR, not introduced by it.
- No `#[repr(C)]`, `Schema` variant, consumer-group, or `state.rs` changes — no STOP-tripwire scope.
- `.config/nextest.toml` OpenSearch test-group addition mirrors the existing Elasticsearch group exactly (serialized, shared reusable container by fixed name) — consistent pattern, not scope creep.

## Simplifications: none (beyond the low-priority dead-branch note above, not worth acting on)

## Verdict: APPROVE

runtime

# PR 3873 review — runtime lane (FFI host, `runtime/src/`)

Scope check first: `git diff --stat` / `files.txt` touch zero files under
`core/connectors/runtime/src/`. Only `runtime/example_config/connectors/opensearch_sink.toml`
(new example, no code). This PR is a new sink plugin
(`core/connectors/sinks/opensearch_sink/`) plus its integration tests, README/registry
entries, and workspace/nextest/bump-version plumbing. Nothing in my ownership
(FFI dispatch, plugin lifecycle, `plugin_id`, `LogCallback`, container `Arc`,
`DashMap`, state atomic-rename, metrics/config plumbing) is modified.

## Verified against runtime source (not changed by this PR)

- PR body claims: `core/connectors/runtime/src/sink.rs:740-748` invokes the FFI
  `consume` callback as a bare statement, discarding its `i32`, so a plugin
  failure never reaches `ConnectorStatus`/`last_error`/`/stats`, and combined
  with `AutoCommit::When(PollingMessages)` (`sink.rs:522`) the offset is
  already committed by the time `consume` runs. Read `sink.rs:480-756`
  directly: confirmed accurate on both counts — `process_messages` calls
  `(consume)(plugin_id, ...)` at `sink.rs:740` with no binding, and
  `setup_sink_consumers` builds the consumer with
  `.auto_commit(AutoCommit::When(AutoCommitWhen::PollingMessages))` at
  `sink.rs:522`. Pre-existing, untouched by this diff, correctly scoped out
  of the PR rather than silently relied on. (pre-untouched, conf:H)
- `opensearch_sink/README.md`'s "Delivery Semantics" section states the same
  gap and additionally claims it verified the failure mode against a live
  server (mapping-conflict batch logged as `PermanentHttpError`, connector
  stayed `Running`). Consistent with the runtime code read above — a
  plugin-level `Err` from `consume()` has no path back into the host's status
  machinery, so "stays Running" is the only possible outcome given current
  `sink.rs`. No fix expected or requested here; this is the same repo-wide
  limitation the PR states affects every sink.

## Connector-mandatory checks, in scope for this plugin

- `SecretString` on `password` (`lib.rs:591`), no `Serialize` derive on
  `OpenSearchSinkConfig`/`ResolvedOpenSearchSinkConfig` — both have hand
  comments explaining why (`lib.rs:584-585`, `lib.rs:651`). `Debug` impls are
  hand-written to redact the URL and never print `password` in plaintext via
  a derived path (`lib.rs:619-630`, `652-675`). Matches `postgres_sink`/
  `http_sink` convention.
- `consume(&self, ...)` (`lib.rs:1211`) — immutable receiver, matches the
  `Sink` trait and every other sink; no interior mutability via `Mutex`, only
  `AtomicU64` counters, so no lock-across-`.await` risk.
- No `tokio::spawn` / `block_on` anywhere in the plugin (grepped the whole
  file) — retries use `tokio::time::sleep`/`tokio::time::timeout` inline in
  the `&self` async fns, which is how the runtime expects a plugin to behave
  under its own executor.
- `[lib] crate-type = ["cdylib", "lib"]` present (`Cargo.toml:221-222`).
- `BTreeMap<HeaderKey, HeaderValue>` used for headers (`lib.rs:1623`,
  matches `ConsumedMessage.headers` shape), converted explicitly to JSON
  since `HeaderKey`/`HeaderValue` aren't `Serialize` as map keys — same
  workaround shape the PR body says `elasticsearch_sink`/`meilisearch_sink`
  skip (verified true against master's `serialize_headers` remark; this sink
  does the conversion by hand rather than reusing that helper, but does not
  drop headers, which is the actual bug in the other two).
- `Vec::with_capacity(messages_count)` at `lib.rs:1251`.
- FFI panic-safety: `documents_at` (`lib.rs:1438-1446`) explicitly drops
  out-of-range positions from a server-echoed `items` array instead of
  indexing/panicking, with a comment citing exactly the right reason — a
  panic across the `extern "C"` plugin boundary aborts the whole connectors
  runtime process. Backed by a dedicated regression test
  (`given_out_of_range_positions_should_drop_them_instead_of_panicking`,
  `lib.rs:2731-2741`). This is the one place in the plugin that actually
  touches a runtime-host invariant, and it's handled correctly.
- Logging: `info!`/`debug!` pair gated on `verbose_logging`
  (`lib.rs:1219-1243`, `1290-1300`) duplicates the exact same format string
  under both branches. Checked whether this is a plugin-specific tell:
  `mongodb_sink/src/lib.rs:263-273` and `redshift_sink/src/lib.rs:417-427` do
  the identical `if self.verbose { info!(...) } else { debug!(...) }`
  duplication already on master. Established repo-wide pattern, not
  introduced here — not flagging as a new simplify target for this PR alone.
  (pre-untouched convention, closest exemplar: `mongodb_sink`)
- Example config path has no platform extension
  (`path = "target/release/libiggy_connector_opensearch_sink"`,
  `example_config/connectors/opensearch_sink.toml:9`) — matches
  `delta_sink.toml`/`clickhouse_sink.toml`/`iceberg_sink.toml`, the runtime
  resolves the `.so`/`.dylib`/`.dll` suffix itself. Correct.

## STOP tripwires

None triggered: no `#[repr(C)]` touched, no `Schema` variant change, no
transient↔`Permanent*` promotion in runtime code (the plugin's own
classification in `is_transient_error`/`map_client_error`/`map_status_error`
is plugin-local, not a runtime reclassification), no consumer-group default
rename (`opensearch_sink_connector` is this plugin's own new default, not a
rename of an existing one), no plugin path resolution change, no
`iggy_connector_sdk` trait change.

## Findings

None in the runtime lane — no `runtime/src/` files changed, and everything
this plugin does that touches a runtime-owned invariant (FFI boundary panic
safety, crate-type, `&self` receiver, no blocking calls, config
forward-compat via `Option<T>`) is handled correctly.

Simplifications: none (no runtime-owned code changed; the one duplication
noted above — verbose/info vs debug logging — is a pre-existing repo-wide
pattern, not something this PR should be asked to fix unilaterally).

Verdict: APPROVE - no runtime/src/ changes; the plugin's own runtime-facing
behavior (FFI panic safety, crate-type, blocking-call discipline, config
forward-compat) is correct, and its two disclosed pre-existing runtime gaps
(consume() return value discarded, offset auto-committed before consume())
were independently verified against current `sink.rs` and are accurately
scoped out of this PR rather than papered over.

sdk

# SDK role review — PR #3873 (opensearch_sink) @ f3d9688d

Scope note: no files under `core/connectors/sdk/src/` are touched by this diff.
Review covers whether the new plugin correctly *uses* the existing SDK
contract (`Sink` trait, `Error` enum, `retry.rs`, `convert.rs`, `sink_connector!`
macro) rather than a contract change. Closest exemplar per PR body:
`elasticsearch_sink` (also compared against `http_sink`, `meilisearch_sink` for
config-field conventions).

## Findings

[nit] opensearch_sink::OpenSearchSinkConfig (core/connectors/sinks/opensearch_sink/src/lib.rs:74-75) - comment "the only in-tree helper for a `SecretString` field writes the credential in plaintext" is factually wrong: `iggy_common::serde_secret::serialize_optional_redacted` (core/common/src/utils/serde_secret.rs:85-93) exists precisely for a struct that must serialize but must not expose the secret, and predates this PR (added before f3d9688d; the module's doc header even warns against exactly this kind of inverted claim after a prior fix, commit f900e7abe "fix the inverted serde_secret redaction claim", #3803). Omitting `Serialize` entirely is a defensible choice on its own merits (elasticsearch_sink instead derives `Serialize` with `serialize_optional_secret`, i.e. plaintext-on-purpose for the control API; clickhouse_sink/delta_sink/doris_sink/redshift_sink also omit `Serialize` like this PR does — split convention, not a deviation), but the justification comment misstates the available tools and could steer a future maintainer wrong. Fix: reword to something like "No `Serialize`: nothing in-tree serializes this type, and `serialize_optional_redacted` is unused here because dropping the impl entirely is simpler than remembering to route password through it." (intro, conf:H)

Simplifications: none. The bulk-retry state machine (`BulkOutcome`/`BulkAttempt`/`index_chunk`) looks large but every branch is exercised by a named test (`given_*`) and each piece (partial-chunk accounting, transient vs permanent classification, first-failure retention) maps to a real OpenSearch `_bulk` behavior called out in the PR body; collapsing it would drop correctness, not just lines. `AtomicU64` counters instead of `elasticsearch_sink`'s `Mutex<State>` is already a simplification (no lock, no lock-across-await concern).

## SDK-contract compliance checklist

- `Sink` trait (core/connectors/sdk/src/lib.rs:133-147): `open(&mut self)`, `consume(&self, ...)`, `close(&mut self)` signatures match exactly; `consume` takes `&self` as required (no interior `Mutex` needed, uses `AtomicU64`).
- `Send + Sync` on `Sink`: satisfied automatically — every field (`AtomicU64`, `Option<OpenSearch>`, `ResolvedOpenSearchSinkConfig` of `String`/`bool`/`Duration`/`Option<Value>`/`Option<Refresh>`/`SecretString`) is `Send + Sync`. No manual impl needed or attempted.
- `sink_connector!(OpenSearchSink)` (lib.rs:53) — plain macro invocation, matches the macro's expansion in core/connectors/sdk/src/sink.rs:239-318 (duplicate-ID guard, `0`/`-1`/`1` FFI codes, `INSTANCES` global, `version()` static) — all handled by the macro, plugin does nothing FFI-level itself. `crate-type = ["cdylib", "lib"]` set correctly in Cargo.toml:32-33.
- `Error` variants used (`InvalidConfigValue`, `Connection`, `InitError`, `HttpRequestFailed`, `InvalidRecordValue`, `PermanentHttpError`, `Serialization`) — all pre-existing variants in core/connectors/sdk/src/lib.rs:389+, used per their documented retry semantics: `HttpRequestFailed` = transient (checked by `is_transient_error`/`retry_on_open`), `PermanentHttpError` = non-retryable data/schema issue (matches the variant's own doc comment about circuit breakers not tripping on bad data). No new `Error` variant added — correct, none of these needed distinct handling.
- `retry.rs` helpers (`exponential_backoff`, `jitter`, `parse_duration`, `is_transient_status`) used with correct signatures. `is_transient_status(status: reqwest::StatusCode)` called with `opensearch::http::StatusCode` — verified same type: `opensearch::http::mod.rs:41` is `pub use reqwest::StatusCode;`, and `reqwest::StatusCode` is itself `pub use http::{StatusCode, ...}` (reqwest-0.12.28/src/lib.rs:279) — no type mismatch, compiles.
- `max_retries` semantics ("N retries after the initial attempt", not "N total attempts") differs from `retry.rs::HttpRetryMiddleware`'s self-documented "total attempt count" convention, but matches the established sink-level convention: http_sink's own doc comment (`batch_length * (max_retries + 1) * ...`, http_sink/src/lib.rs:1191) uses the same "+1 for initial attempt" semantics. Not a deviation — two different sub-conventions coexist pre-PR (middleware-level vs bespoke-retry-loop sinks), and this plugin correctly follows the bespoke-loop one it actually implements.
- `convert::owned_value_to_serde_json` (core/connectors/sdk/src/convert.rs:27) called with `&simd_json::OwnedValue` from a destructured `Payload::Json` — signature matches, same usage pattern as elasticsearch_sink.
- `Payload` match in `prepare_document` (lib.rs:410-423) handles `Json`/`Raw`/`Text` explicitly, `Proto`/`FlatBuffer`/`Avro` fall to an explicit `Error::InvalidRecordValue` (not a silent drop) — acceptable scope limit, no SDK contract violation.
- `SecretString` on `password` (config field, lib.rs:81) with `ExposeSecret` used only inside `create_client` (Basic auth header) — never logged, never in a `Display`/`Debug` path (manual `Debug` impls on both `OpenSearchSink` and `ResolvedOpenSearchSinkConfig` explicitly avoid deriving through the client/password, with an accurate comment explaining why on lib.rs:105-108 and lib.rs:141).
- No `tokio::spawn` / `block_on` / `std::sync::Mutex` anywhere in the plugin (verified by grep across lib.rs); all `.unwrap()`/`.expect()` calls are confined to `#[cfg(test)] mod tests` (first non-test line is 1351, all unwrap/expect hits are ≥1405).
- BDD test naming: all unit tests are `#[test] fn given_..._should_...` — consistent 3-part naming, matches CLAUDE.md convention.
- Config forward-compat: every `OpenSearchSinkConfig` field is `Option<T>` except the two required ones (`url`, `index`), consistent with `#[serde(default)]`-equivalent forward compatibility (no `#[serde(default)]` needed since there's no `Default` derive requirement beyond what's already `Option`-wrapped; struct does derive `Default`).
- `Refresh` config field: `Option<Refresh>` where `opensearch::params::Refresh` derives `Deserialize` with `#[serde(rename = "true"/"false"/"wait_for")]` (opensearch-2.4.0/src/params.rs:152-159) — `refresh = "false"` in config.toml/README/example_config all deserialize correctly.
- Workspace `Cargo.toml` diff (lines 89-101 of diff.patch) only adds the new workspace member and the `opensearch = "2.4.0"` dependency — no version bump to `iggy_connector_sdk` or any STOP-tripwire crate in this PR's actual diff (the version-bump lines visible via a stale `git diff HEAD~1` are from the unrelated `Merge branch 'master'` commit at HEAD, not from this PR's content — confirmed against diff.patch directly).

## Verdict: APPROVE - clean SDK-contract usage; one nit-level comment inaccuracy (not a correctness or security issue — the credential itself is still never leaked, since `Serialize` is correctly omitted either way)

testing

# Testing review - PR #3873 (opensearch_sink)

[nit] core/integration/tests/connectors/opensearch/opensearch_sink.rs::given_json_messages_when_sink_consumes_should_index_documents_and_upsert_by_natural_key (opensearch_sink.rs:33) - integration test uses unit-test BDD `given_when_should` naming instead of `connector-testing` SKILL.md's mandated declarative `<subject>_<action>_<observation>` for integration tests ("never given_should there"). Same pattern in opensearch_sink_failures.rs:132 (`given_missing_index_and_mapping_conflict_should_isolate_failures_from_healthy_sibling`). Fix: rename to e.g. `json_messages_sink_indexes_and_upserts_by_natural_key` / `missing_index_and_mapping_conflict_isolate_from_healthy_sibling`, matching `postgres_sink.rs::json_messages_sink_stores_as_bytea` or `elasticsearch_sink.rs::elasticsearch_sink_stores_json_messages`. Low priority: `meilisearch_sink.rs::given_json_messages_when_sink_consumes_should_index_documents` (near-identical name/shape to this PR's test) and `doris_sink.rs`, `quickwit_sink.rs`, `runtime/benchmark.rs`, `runtime/http_state.rs` all already use the same `given_should` style in integration tests, so this is repo-wide drift the PR inherited rather than introduced. (origin: pre-surfaced, conf: H)

Simplifications: none. Unit test file (`core/connectors/sinks/opensearch_sink/src/lib.rs:1351-3085`, ~1730 lines) is large but every test targets a distinct branch with a comment justifying why (e.g. the four separate `given_debug_formatted_*_should_redact_password` tests each catch a different leak surface: config Debug, sink Debug with no client, sink Debug with a real opened `OpenSearch` client - which has its own un-redacted `Debug` derive down through `Credentials::Basic` - and URL-embedded credentials). No duplicate/mocked-should-be-real-infra tests, no fixed sleeps outside legitimate poll loops or backoff-under-test.

## What was verified

- **BDD naming, unit tests**: consistent `given_X_should_Y` / `given_X_when_Y_should_Z` throughout `opensearch_sink/src/lib.rs::tests` (lib.rs:1408-3084). No `test_foo`/`does_bar` names.
- **Config helper**: `base_config()` (lib.rs:1386) matches the naming convention of the closest exemplar `meilisearch_sink::tests::base_config` (not `test_config`, which is fine - `meilisearch_sink`, `mongodb_sink` (`given_default_config`), and `http_sink` (`given_default_config`) all name this helper differently; no single fixed name is enforced repo-wide).
- **`#[tokio::test]` in `src/lib.rs`**: used for all async unit tests (lib.rs:2241 onward) rather than local `Runtime::new()`. This looks like a skill-guideline deviation at first read, but the closest exemplars for wiremock-backed sink tests, `http_sink/src/lib.rs` and `mongodb_sink/src/lib.rs`, both already use `#[tokio::test]` directly in `src/lib.rs`. The `Runtime::new()` guidance in the skill is written for source state round-trip tests (`random_source`-style); sinks with real async HTTP mocking already establish `#[tokio::test]` as the norm. Not a defect. (origin: pre-surfaced, conf: H)
- **Sink pure-logic coverage** (mandatory checklist item): config defaults + fallback/clamping (lib.rs:1409-1442), all three `Payload` variants incl. non-object JSON wrapping and raw-non-JSON base64 fallback (lib.rs:1494-1715), header encoding incl. the dynamic-mapping-collision regression test (lib.rs:1648-1684), `_bulk` response/query building and parsing incl. malformed/short/missing-items edge cases (lib.rs:1923-2231), and transient-vs-permanent classification for both client errors and per-item bulk statuses (lib.rs:2845-2856, 1170-1195, plus the wiremock-driven retry tests at lib.rs:2394-2653). This is the most thorough sink test suite of the set I compared against (`elasticsearch_sink`, `meilisearch_sink`, `mongodb_sink`, `http_sink`).
- **No four canonical source-state tests required**: this is a sink, not a source - correctly out of scope.
- **Integration layout**: `core/integration/tests/connectors/opensearch/{mod.rs,opensearch_sink.rs,opensearch_sink_failures.rs,sink.toml,failure_states.toml,failure_states/*.toml}` plus `fixtures/opensearch/{mod.rs,container.rs,sink.rs,failure.rs}` mirrors `postgres`/`elasticsearch` exactly: `TestFixture` impl, `ConfigEnv` env-injection (all three documented forms: plugin-config leaf, indexed `streams_N_*`, top-level `path`), `iggy-test-opensearch` fixed name + `ReuseDirective::Always` matching the `elasticsearch`/`doris` reuse pattern (skill's "as of now only elasticsearch and doris share a container" note is now stale text, not a PR defect), polling helpers (`wait_for_document_count`, `wait_for_status`) instead of fixed sleeps, `iggy-test-` container prefix honored.
- **`.config/nextest.toml`**: new `[test-groups.opensearch]` + `max-threads = 1` override correctly serializes the reused-container test group, matching the `elasticsearch`/`doris` blocks immediately above it (nextest.toml:56-69).
- **Two integration tests cover exactly what's claimed**: happy path (natural-key upsert idempotency proven end-to-end by resending `order_id: A-1` at a new offset, headers round-trip, both `Payload::Raw` branches via a second stream/topic on the `raw` schema) and a real failure-isolation test (missing-index `open()` failure vs. a live `mapper_parsing_exception` mid-`consume()`, three-sink isolation, chunk-boundary survival cross-checked against the unit-level `given_permanently_failing_chunk_should_not_abandon_later_chunks`). The module doc in `opensearch_sink_failures.rs:18-65` explicitly traces the runtime code path (`sink.rs::process_messages` discarding the FFI `consume` return) backing its "not visible anywhere" claim rather than asserting it blind.
- **Doc/config sync**: `opensearch_sink/README.md` config table, `opensearch_sink/config.toml`, and `runtime/example_config/connectors/opensearch_sink.toml` all list the identical `[plugin_config]` field set (defaults match `OpenSearchSinkConfig`'s `unwrap_or` values in lib.rs). `core/connectors/README.md` and `core/connectors/sinks/README.md` both gained the new sink row in alphabetical position. No `max_connections` field applies here (sink has no connection pool); retry knobs (`max_retries`, `retry_delay`, `max_retry_delay`, `max_open_retries`) and `verbose_logging` are all present in both TOMLs and documented.
- **Investigated and ruled out**: an apparent "PR deletes `rabbitmq_sink`" signal from a raw `git diff master -- ...` was a false positive from the local `master` ref having advanced past this PR's actual merge-base (post-merge commits landed on master after this branch's base). The actual `diff.patch` supplied for review contains zero references to `rabbitmq_sink`; confirmed via direct grep. Not a real finding.

Verdict: APPROVE

Validation record

• Shard validator (validate-1.md): 5/5 claims PASS on substance; 2 FIX (C2 line-anchor correction, C5 origin-tag correction); 0 REMOVE.
• Sweep validator (sweep.md): 5 additions (S1-S3 nit, S4-S5 simplify), all conf H/M, none critical/warning; 0 corrections that reverse a PASS (C5's "repo-wide drift" framing narrowed to "one prior instance, widened by this PR" - doesn't change verdict or severity).
• Contested items: 0 (no REMOVE, no critical/warning downgrade, no sweep addition contradicts a PASS).

[mattp5657]

@slbotbm Please see above.

[slbotbm]

/ready

[hubcio]

verdict: request changes. the mechanics are copied from the team-review that #4095 (this branch's base) replaced, the no-argument path now diffs tip to tip, and the role blocks restate connector rules that master already reversed on 09-14 (#3957) or fixed in the runtime (#4152). rebase, take the current team-review charter, and point the experts at the connector-* skills instead of restating them.

lines that go stale after the rebase: 31 (master logs and counts the consume return, the offset is still committed first, so no redelivery either way), 66 and 82 (retry_async / RetryPolicy / retry_backoff, and meilisearch_sink is the max_retries exception), 68 (the cursor is staged in pending and committed on ack), 85 (the third state test also has a refuse-to-open shape).

sentences copied here that are wrong in the source skills as well, fix there or drop them here: connector-runtime 136-137 and 237 (the source loop rejects the whole batch, your line 75 is the correct one), 147 (flush condition inverted), 181 (the config provider retries 5xx, 408, 429 and transport errors), 247 (LOG_CALLBACK is a const and stats::SYSINFO is the missing static); connector-sdk 80 (no schema list exists in core/connectors/README.md), 141 (1 also means consume failure); connector-source 83 (origin_timestamp is never forwarded); connector-testing 74 and 353 (#[tokio::test] rationale), 177, 312 and 368 (restart.rs), 373 (exemplar paths point at files without tests).

[hubcio]

warning: with no argument this now runs the two-dot git diff origin/master..HEAD, a tip-to-tip diff that drags in every master-side change when the branch is behind master (783 files on this checkout instead of 2). keep the merge-base form for the empty case, or use origin/master...HEAD.

[hubcio]

warning: this is the caveman charter that #4095, this branch's own base commit, replaced with the plain english one because caveman is hard to read. rebase the copy on the current team-review (or team-review-slim) and keep only the connector deltas.

[hubcio]

warning: master's connector-source skill now says the opposite of this line: return state: None on an empty poll unless a watermark advanced. restating the connector skills here drifts within days, so have each expert load its connector-* skill via the Skill tool and keep this block to focus areas.

also at line 68: on master a state serialization failure is a poll error, not non-fatal.

[hubcio]

warning: the sink loop flushes when the message is the polled partition head or the batch is full (runtime/src/sink.rs:352), and a gap keeps accumulating, so "flush on offset gap" is inverted. the connector-runtime skill has the same wording at line 147, fix both.

[hubcio]

warning: 32 of 169 harness tests under tests/connectors/ use given_ names, six files entirely, and AGENTS.md's own filter example is given_logging_format_json. drop "never" and flag only mixed styles inside one file, as the testing skill says.

[hubcio]

nit: 1 is also what the macro returns when consume fails (sdk/src/sink.rs:229), and a failed close is logged and returns 0. say "1 = plugin-reported failure from open or consume".

[hubcio]

nit: only avro has try_new, proto's new is lenient without one and flatbuffer's cannot fail. restore the sdk skill's scope: try_new only where schema loading can fail.

[hubcio]

nit: the runtime never forwards origin_timestamp (build_iggy_message takes payload, id and headers only) and the server stamps micros, so a nanos check is noise on a discarded field. say the runtime drops it today.

[hubcio]

nit: worth adding that Err from on_batch_result stops the source (sdk/src/source.rs:518-521), so transient backend failures must be retried inside the callback. a plugin reviewer would pass that today.

[hubcio]

simplification: on a plugin-only diff (no runtime/src or sdk/src in files.txt) the runtime and sdk experts own nothing, and in the #3873 run both said so and produced one nit between them. merge them into one contracts expert for that case.