fix(#1425): strict write gate no longer consumes one-shot row iterables

The strict-provenance write gate ran assert_write_allowed(self, rows) before
insert materialized rows, and its key-consistency check did 'for row in rows'.
For a one-shot iterable (generator, map, iter), that exhausted it, so the
downstream insert saw an empty iterator and wrote zero rows with no error —
silent data loss on compliant self.insert(<generator>) code (common for Part
inserts). insert(<query_expression>) was also double-executed.

Split the gate: assert_write_allowed(target) now does the target-only check
(needs no rows) in Table.insert; a new per-row assert_row_key_allowed(row)
runs in Table._insert_rows as each row is materialized — the single point
reached by both the chunked and single-batch paths, so streaming/chunking is
preserved and the caller's iterable is never consumed early. The
QueryExpression (INSERT ... SELECT) path is no longer iterated by the gate,
fixing the double-execution; per-row key consistency does not apply there
(rows never materialize client-side), governed by the target check only.

Adds regression tests: a generator insert into a Part must land all rows, and
the per-row key check still fires for generator-sourced rows.

Fixes the blocking bug flagged by @ttngu207 in review. Read-gate coverage
(len/bool/in, restriction-by-table) is tracked separately pending the
best-effort-vs-close decision.

Author: dimitri-yatsenko

src/datajoint/provenance.py

@@ -4,8 +4,11 @@
 When the flag is enabled, this module's context (set by ``AutoPopulate._populate_one``)
 tracks which tables and primary key the currently-executing ``make()`` is
 allowed to read and write. The read gate in :func:`assert_read_allowed`
-fires inside ``QueryExpression.cursor``; the write gate in
-:func:`assert_write_allowed` fires inside ``Table.insert``.
+fires inside ``QueryExpression.cursor``. The write gate has two parts: the
+target check in :func:`assert_write_allowed` fires inside ``Table.insert``
+(before rows are materialized), and the per-row key-consistency check in
+:func:`assert_row_key_allowed` fires inside ``Table._insert_rows`` as each row
+is materialized — so the gate never consumes the caller's ``rows`` iterable.
 
 The contract is documented in
 ``datajoint-docs/src/reference/specs/provenance.md`` §3.
@@ -122,29 +125,30 @@ def assert_read_allowed(query_expression) -> None:
         )
 
 
-def assert_write_allowed(target_table, rows) -> None:
+def assert_write_allowed(target_table) -> None:
     """
-    Verify an insert is allowed under the active strict-make context.
+    Verify the *target* of an insert is allowed under the active strict-make context.
 
-    Called from ``Table.insert`` after the existing ``_allow_insert`` check.
-    No-op when no strict-make context is active.
+    Called from ``Table.insert`` after the existing ``_allow_insert`` check and
+    before any rows are materialized. No-op when no strict-make context is active.
 
-    Allowed writes:
+    Allowed targets:
 
-    - Target is the current ``make()`` target (``self``) or one of its Part
-      tables.
-    - Every row's primary-key columns that overlap with the current ``key``
-      must equal ``key``'s values.
+    - The current ``make()`` target (``self``) or one of its Part tables.
 
-    Anything else raises ``DataJointError``.
+    Per-row key consistency is checked separately by :func:`assert_row_key_allowed`
+    as rows are materialized, so this gate never consumes the caller's ``rows``
+    iterable — a one-shot generator must survive to reach ``insert``.
+
+    Raises ``DataJointError`` if the target is not permitted.
     """
     ctx = _active_strict_make.get()
     if ctx is None:
         return
 
-    make_target, _allowed_tables, key = ctx
+    make_target, _allowed_tables, _key = ctx
 
-    # 1. Target must be `make_target` (self) or one of its Parts.
+    # Target must be `make_target` (self) or one of its Parts.
     target_name = getattr(target_table, "full_table_name", None)
     target_set = {make_target.full_table_name}
     # Collect Part tables of make_target via class __dict__ (not dir/getattr,
@@ -169,17 +173,26 @@ def assert_write_allowed(target_table, rows) -> None:
             f"table and its Part tables may be written."
         )
 
-    # 2. Each row's key columns that overlap with the current key must match.
-    if isinstance(rows, dict):
-        _check_row_key(rows, key)
-    else:
-        try:
-            for row in rows:
-                if isinstance(row, dict):
-                    _check_row_key(row, key)
-                # Non-dict rows (tuples, etc.) bypass — older API; can't check.
-        except TypeError:
-            pass  # not iterable; let downstream code handle
+
+def assert_row_key_allowed(row) -> None:
+    """
+    Verify a single insert row's key columns match the active ``make()`` key.
+
+    Called per row from ``Table._insert_rows`` as rows are materialized, so the
+    check sees a concrete row without the write gate having to consume the
+    caller's ``rows`` iterable. No-op when no strict-make context is active or
+    when ``row`` is not a dict (numpy records / bare sequences carry no field
+    names to check by — same as the previous behavior).
+
+    Raises ``DataJointError`` on a mismatch.
+    """
+    ctx = _active_strict_make.get()
+    if ctx is None:
+        return
+    if not isinstance(row, dict):
+        return
+    _make_target, _allowed_tables, key = ctx
+    _check_row_key(row, key)
 
 
 def _check_row_key(row: dict, current_key: dict) -> None:

src/datajoint/table.py

@@ -797,16 +797,23 @@ def insert(
                 " To override, set keyword argument allow_direct_insert=True."
             )
 
-        # Strict-provenance write gate. No-op outside make() or when the
-        # config flag is off. See src/datajoint/provenance.py.
+        # Strict-provenance write gate (target check only). No-op outside make()
+        # or when the config flag is off. Deliberately does NOT touch `rows` —
+        # the per-row key-consistency check happens in `_insert_rows` as rows are
+        # materialized, so a one-shot iterable (generator) is not consumed here.
+        # See src/datajoint/provenance.py.
         from .provenance import assert_write_allowed
 
-        assert_write_allowed(self, rows)
+        assert_write_allowed(self)
 
         if inspect.isclass(rows) and issubclass(rows, QueryExpression):
             rows = rows()  # instantiate if a class
         if isinstance(rows, QueryExpression):
-            # insert from select - chunk_size not applicable
+            # insert from select - chunk_size not applicable.
+            # Note: this INSERT ... SELECT runs entirely server-side, so under
+            # strict_provenance the per-row key-consistency check does not apply
+            # (row values are never materialized client-side). The target check
+            # in assert_write_allowed above still governs which table is written.
             if chunk_size is not None:
                 raise DataJointError("chunk_size is not supported for QueryExpression inserts")
             if not ignore_extra_fields:
@@ -861,7 +868,17 @@ def _insert_rows(self, rows, replace, skip_duplicates, ignore_extra_fields):
         """
         # collects the field list from first row (passed by reference)
         field_list = []
-        rows = list(self.__make_row_to_insert(row, field_list, ignore_extra_fields) for row in rows)
+        # Strict-provenance per-row key check runs here, as each row is
+        # materialized — no-op outside make()/when the flag is off. Placing it in
+        # this single materialization point (reached by both the chunked and
+        # single-batch paths) avoids consuming the caller's `rows` iterable early.
+        from .provenance import assert_row_key_allowed
+
+        def _make_row(row):
+            assert_row_key_allowed(row)
+            return self.__make_row_to_insert(row, field_list, ignore_extra_fields)
+
+        rows = list(_make_row(row) for row in rows)
         if rows:
             try:
                 # Handle empty field_list (all-defaults insert)

tests/integration/test_strict_provenance.py

@@ -212,6 +212,84 @@ def make(self, key):
     assert len(Master.Bin & {"subject_id": 1}) == 3
 
 
+def test_strict_generator_insert_not_dropped(prefix, connection_test, strict_mode):
+    """Regression (#1474 bug 1): a one-shot generator of compliant rows must not
+    be consumed by the write gate. Before the fix, assert_write_allowed iterated
+    `rows` for its key check, exhausting the generator so insert saw zero rows and
+    silently wrote nothing."""
+    schema = dj.Schema(f"{prefix}_strict_generator", connection=connection_test)
+
+    @schema
+    class Subject(dj.Lookup):
+        definition = """
+        subject_id : int32
+        """
+        contents = [(1,), (2,)]
+
+    @schema
+    class Spectrum(dj.Computed):
+        definition = """
+        -> Subject
+        ---
+        n : int32
+        """
+
+        class Bin(dj.Part):
+            definition = """
+            -> master
+            bin_id : int32
+            ---
+            energy : float64
+            """
+
+        def make(self, key):
+            n = 5
+            self.insert1({**key, "n": n})
+            # one-shot generator (not a list) — must survive the write gate
+            self.Bin.insert({**key, "bin_id": i, "energy": float(i)} for i in range(n))
+
+    Spectrum.populate()
+    for sid in (1, 2):
+        assert (Spectrum & {"subject_id": sid}).fetch1("n") == 5
+        # The core assertion: all 5 generated rows landed, none silently dropped.
+        assert len(Spectrum.Bin & {"subject_id": sid}) == 5
+
+
+def test_strict_generator_insert_mismatched_key_still_caught(prefix, connection_test, strict_mode):
+    """The per-row key check still fires when rows come from a generator — a row
+    whose key disagrees with the current make() key raises, not silently passes."""
+    schema = dj.Schema(f"{prefix}_strict_gen_mismatch", connection=connection_test)
+
+    @schema
+    class Subject(dj.Lookup):
+        definition = """
+        subject_id : int32
+        """
+        contents = [(1,)]
+
+    @schema
+    class Derived(dj.Computed):
+        definition = """
+        -> Subject
+        ---
+        val : int32
+        """
+
+        class Bin(dj.Part):
+            definition = """
+            -> master
+            bin_id : int32
+            """
+
+        def make(self, key):
+            self.insert1({**key, "val": 0})
+            # generator whose 3rd row carries a bogus subject_id
+            self.Bin.insert({**({**key, "subject_id": 999} if i == 2 else key), "bin_id": i} for i in range(4))
+
+    with pytest.raises(DataJointError, match="does not match the current make"):
+        Derived.populate()
+
+
 def test_strict_off_by_default_no_change(prefix, connection_test):
     """With strict_provenance unset (default False), existing patterns work unchanged."""
     schema = dj.Schema(f"{prefix}_strict_default_off", connection=connection_test)