#3775 First SIGTERM during a normal target drain can re-enter a sink and acknowledge uncommitted data Issue created by SBrouck Observed behavior On SDK main

43b3563734afce61a067ad0233b6720ea8c4b455

, a first SIGTERM received during an ordinary

Target.drain_all()

can re-enter

process_batch

for the same sink. With a minimal SQLite BatchSink that tolerates an empty record collection, the target emits bookmark 5 and exits with code 0 while no rows have been committed. Restarting from that checkpoint with inclusive replication loads only ID 5, leaving IDs 0–4 absent. This is distinct from #3698: that change protects a shutdown already in progress. Here

_is_terminating

is false when the first signal arrives during normal draining. Mechanism

Sink.start_drain()

moves

_pending_batch

into

_context_draining

and clears

_pending_batch

, but the record count stays positive until

mark_drained()

. The signal handler immediately invokes another

drain_all()

. The nested

start_drain()

replaces the in-flight context with

{}

. If processing that empty context succeeds, the nested drain emits

_latest_state

and exits before the original batch commits. This outcome depends on the sink. The SDK's reference CSVSink requires `context["records"]`: in the same instrumented timing window it instead encounters an empty context, stops without completing the original write, and emits no checkpoint. I am not claiming that every destination acknowledges missing rows. Minimal reproduction Run the following in a fresh Python process with singer-sdk installed at the revision above. It sends SIGTERM only to itself. Catching SystemExit at the outermost level lets the script display the captured target output and database contents after the original call stack has unwound. """Run in a fresh Python process; sends SIGTERM only to that process.""" from future import annotations import contextlib import io import json import os import signal import sqlite3 from singer_sdk.sinks import BatchSink from singer_sdk.target_base import Target class SQLiteSink(BatchSink): def __init__(self, target, **kwargs): self.owner = target super().__init__(target=target, **kwargs) def process_batch(self, context): rows = context.get("records", []) if not self.owner.signalled: self.owner.signalled = True os.kill(os.getpid(), signal.SIGTERM) with self.owner.db: self.owner.db.executemany( "INSERT INTO rows VALUES (?)", [(r["id"],) for r in rows] ) class SQLiteTarget(Target): name = "target-signal-example" default_sink_class = SQLiteSink def __init__(self): self.db = sqlite3.connect("memory") self.db.execute("CREATE TABLE rows (id INTEGER PRIMARY KEY)") self.signalled = False super().__init__() self._max_parallelism = 1 messages = [ { "type": "SCHEMA", "stream": "rows", "schema": {"type": "object", "properties": {"id": {"type": "integer"}}}, "key_properties": ["id"], }, *[{"type": "RECORD", "stream": "rows", "record": {"id": i}} for i in range(6)], { "type": "STATE", "value": { "bookmarks": {"rows": {"replication_key": "id", "replication_key_value": 5}} }, }, ] target = SQLiteTarget() output = io.StringIO() with contextlib.redirect_stdout(output): try: target.listen(io.StringIO("".join(json.dumps(m) + "\n" for m in messages))) except SystemExit as exc: exit_code = exc.code print( json.dumps( { "exit_code": exit_code, "acknowledged_states": output.getvalue().splitlines(), "persisted_rows": target.db.execute("SELECT id FROM rows").fetchall(), }, indent=2, ) ) Observed:

exit_code=0

, an acknowledged state with replication_key_value 5, and

persisted_rows=[]

. Validation and possible direction Local environment: macOS ARM64, Python 3.12.4. A separate file-backed SQLite subprocess experiment verifies the missing IDs after restart. The incomplete acknowledgement repeated in 10/10 runs (five with one worker, five with two). Additional controls include the SDK CSV sink, signals before/after commit, and duplicate signals during shutdown. A diagnostic in-memory intervention deferring the first signal until the outer normal drain completes removes the incomplete acknowledgement in the corresponding ten runs. Eight targeted behavioral checks fail before and pass with that intervention; 82 existing targeted tests also pass with it. This is evidence for the mechanism, not a production-ready patch or a full-suite validation. Would deferring shutdown until the active drain completes be the appropriate design? It needs to cover both direct size-triggered drains and global drains, avoid duplicate cleanup, and preserve failures. A separate existing behavior still masks commit errors during an already-started shutdown via `finally: _terminate()`; the intervention does not claim to fix that. I noticed #3757 changes when batches drain, so coordination may be useful. No upstream files have been changed. I used Codex to assist for this investigation, reproductions, diagnostic intervention and local validation. meltano/sdk