fix(storage): probe before the pipe, not inside it (#462)

scripts/drivers/storage/sqlite-sync.sh

@@ -62,6 +62,7 @@ _SQLITE_SYNC_COMMIT_BYTES=131072
 _sqlite_sync_commit_chunk() {
   local db="$1" sql="$2"
   [ -n "$sql" ] || return 0
+  agmsg_sqlite_warm
   printf 'BEGIN IMMEDIATE;\n%s\nCOMMIT;\n' "$sql" | agmsg_sqlite -bail "$db" >/dev/null 2>&1
 }
 
@@ -867,7 +868,7 @@ storage_sync_reconcile_push() {
 
   # Stdin, for the same reason as the pull outcomes (#882): `$values` gains an
   # entry per acked message and a full catch-up push carries a thousand.
-  printf '%s\n' "BEGIN IMMEDIATE;
+  _sqlite_exec_stdin "$db" "BEGIN IMMEDIATE;
     CREATE TEMP TABLE incoming_sync_acks(
       local_position INTEGER UNIQUE,wire_id TEXT UNIQUE,server_seq TEXT UNIQUE);
     INSERT INTO incoming_sync_acks VALUES $values;
@@ -939,7 +940,7 @@ storage_sync_reconcile_push() {
     WHERE b.local_team='$tl' AND b.server_instance_id='$server'
       AND b.remote_team_id='$remote' AND b.protocol_version=$protocol
       AND b.driver_generation='$generation';
-    COMMIT;" | agmsg_sqlite -bail -batch "$db" >/dev/null 2>&1 || return 12
+    COMMIT;" >/dev/null 2>&1 || return 12
 
   _sqlite_data "$team" "SELECT json_object('type','sync_reconcile_result','push_cursor',
     CAST(push_cursor AS TEXT)) FROM sync_bindings WHERE local_team='$tl'
@@ -1434,7 +1435,7 @@ EOF
 
   # Stdin, third of the same kind (#882): `$insert_members` carries one row per
   # roster member and `$insert_local_agents` one per local agent.
-  printf '%s\n' "BEGIN IMMEDIATE;
+  _sqlite_exec_stdin "$db" "BEGIN IMMEDIATE;
     CREATE TEMP TABLE incoming_read_members(member_id TEXT UNIQUE,agent TEXT UNIQUE);
     CREATE TEMP TABLE local_read_agents(agent TEXT PRIMARY KEY);
     $insert_members
@@ -1529,7 +1530,7 @@ EOF
        AND rm.remote_team_id='$remote' AND rm.protocol_version=$protocol
        AND rm.driver_generation='$generation' AND rm.active=1
        AND rm.name_mismatch=0;
-    COMMIT;" | agmsg_sqlite -bail -batch "$db" >/dev/null || { _sqlite_sync_why; return 13; }
+    COMMIT;" >/dev/null || { _sqlite_sync_why; return 13; }
 
   _sqlite_data "$team" "SELECT json_object('type','sync_read_frontier','member_id',f.member_id,
       'server_seq',f.server_seq) FROM sync_read_prepared f JOIN sync_read_members rm

scripts/drivers/storage/sqlite.sh

@@ -57,9 +57,22 @@ _sqlite_data() {
 # reading a non-tty is still willing to treat a malformed line as an
 # interactive prompt, and the point of this path is that nobody is watching.
 _sqlite_data_stdin() {
+  # Outside the subshell on purpose: a probe run inside it would be discarded.
+  agmsg_sqlite_warm
   ( set -o pipefail; printf '%s\n' "$2" | agmsg_sqlite -batch "$(_sqlite_db "$1")" | tr -d '\r' )
 }
 
+# The same, for a statement whose output nobody reads. Takes a database PATH
+# rather than a team, because its callers are inside the driver and hold one.
+# -bail as at the two driver sites this replaced: the stdin form must stop at
+# the first error so a busy call has written nothing of a transaction that
+# never began, which is what lets the engine retry it. The warm call sits on
+# the line above the pipe, where the #462 scan looks for it.
+_sqlite_exec_stdin() {
+  agmsg_sqlite_warm
+  printf '%s\n' "$2" | agmsg_sqlite -bail -batch "$1"
+}
+
 # IN (...) list of "team:agent" pairs.
 _sqlite_pair_in() {
   local out="" p t a
@@ -276,8 +289,10 @@ storage_send() {
   # inserted the message a second time, leaving one row in the legacy table that
   # no event points at -- exactly the unlinked copy the correspondence exists to
   # prevent.
+  agmsg_sqlite_warm
   if ! printf '%s\n' "$insert" | agmsg_sqlite -bail "$db" >/dev/null 2>&1; then
     storage_init "$team" >/dev/null
+    agmsg_sqlite_warm
     printf '%s\n' "$insert" | agmsg_sqlite -bail "$db" >/dev/null 2>&1 || return 1
   fi
   printf '%s\n' "$id"
@@ -521,6 +536,7 @@ storage_import() {
       frm=$(j from); to=$(j to); body=$(j body)
       # Same utility as a live send, so an imported store presents the same
       # legacy view as the store it came from (#689).
+      agmsg_sqlite_warm
       printf '%s\n' "$(_sqlite_message_sent_sql "$team" "$frm" "$to" "$body" "$id" "$at")" \
         | agmsg_sqlite -bail "$db" >/dev/null 2>&1
     elif [ "$t" = message_read ]; then

scripts/internal/migrate-team-store.sh

@@ -54,6 +54,7 @@ _drop_from_shared() {
     sql="$sql DELETE FROM $t WHERE team='$lit';"
   done
   sql="$sql COMMIT;"
+  agmsg_sqlite_warm
   printf '%s\n' "$sql" | agmsg_sqlite "$SHARED" >/dev/null
 }
 
@@ -178,6 +179,7 @@ _missing_from_dest() {
     # A destination that cannot be read, or lacks the table, makes the query
     # fail — which is reported as "not proven complete", never as "nothing is
     # missing". Being unable to check is not the same as having checked.
+    agmsg_sqlite_warm
     out="$(printf '%s\n' "ATTACH DATABASE '$dest_lit' AS dst; $sql" \
       | agmsg_sqlite "$SHARED" 2>/dev/null)" || { echo "$t"; return 0; }
     [ -z "$out" ] || { echo "$t"; return 0; }
@@ -340,6 +342,7 @@ fi
 copy="$copy
   COMMIT;"
 
+agmsg_sqlite_warm
 printf '%s\n' "ATTACH DATABASE '$src_lit' AS src;
 $copy" | agmsg_sqlite "$DEST" >/dev/null
 

scripts/lib/storage.sh

@@ -225,6 +225,22 @@ _agmsg_escape_flag() {
   printf '%s' "$_AGMSG_ESCAPE_FLAG"
 }
 
+# Run the escape probe in THIS shell, before a pipeline starts.
+#
+# `agmsg_sqlite` memoises the probe so it costs one sqlite3 process per shell
+# rather than one per call (#462). The right-hand side of a pipeline is a
+# subshell: it inherits the memo, but a memo it sets there dies with it. So a
+# process whose FIRST database access is piped records nothing, and every piped
+# call after it probes again -- measured at two sqlite3 processes per call, and
+# it never converges.
+#
+# A REDIRECTION IS NOT A PIPE. `agmsg_sqlite db < file` runs in the current
+# shell and memoises normally; only `... | agmsg_sqlite ...` needs this. Call it
+# on the line before the pipeline, not inside it.
+agmsg_sqlite_warm() {
+  [ -n "$_AGMSG_ESCAPE_PROBED" ] || _agmsg_escape_flag >/dev/null
+}
+
 agmsg_sqlite() {
   # Probe in THIS shell, not in a command substitution. `$(_agmsg_escape_flag)`
   # ran the function in a subshell, so the memo it set was discarded on exit and

tests/test_remote_sync.bats

@@ -1265,3 +1265,79 @@ _longest_argv() {
   prepare_push >/dev/null
   [ "$(sqlite3 "$db" "SELECT COUNT(*) FROM sqlite_master WHERE type='index' AND name='sync_messages_server_seq';" | tr -d '\r')" -eq 1 ]
 }
+
+# Piped `agmsg_sqlite` sites under <root>/scripts with no warm within the two
+# lines above the pipe, one "file:line" per line. Comment lines are not sites:
+# the rule is about processes, and a comment starts none. A whole file is never
+# excluded -- a real pipeline added to any file must be seen (review finding on
+# #904: the earlier file-level skip would have hidden one).
+_462_unwarmed_under() {
+  local root="$1" file n line stripped prev
+  while IFS=: read -r file n line; do
+    stripped="${line#"${line%%[![:space:]]*}"}"
+    case "$stripped" in \#*) continue ;; esac
+    prev="$(sed -n "$((n > 2 ? n - 2 : 1)),$((n))p" "$root/$file")"
+    case "$prev" in *agmsg_sqlite_warm*) continue ;; esac
+    printf '%s:%s\n' "$file" "$n"
+  done < <(cd "$root" && grep -rn '| agmsg_sqlite' scripts/)
+}
+
+# Executable (non-comment) piped sites under <root>/scripts: the population
+# the scan judges, counted the same way the scan reads them.
+_462_piped_sites_under() {
+  local root="$1" file n line stripped count=0
+  while IFS=: read -r file n line; do
+    stripped="${line#"${line%%[![:space:]]*}"}"
+    case "$stripped" in \#*) continue ;; esac
+    count=$((count + 1))
+  done < <(cd "$root" && grep -rn '| agmsg_sqlite' scripts/)
+  printf '%s\n' "$count"
+}
+
+@test "storage: every piped agmsg_sqlite warms the escape probe first (#462)" {
+  # THE SET IS DERIVED, NOT LISTED. `agmsg_sqlite` memoises the escape probe so
+  # it costs one sqlite3 process per shell rather than one per call, and the
+  # right-hand side of a pipeline is a subshell: it inherits a memo but cannot
+  # leave one behind. A process whose first database access is piped therefore
+  # probes on every call, forever -- the cost #462 removed. A redirection
+  # (`agmsg_sqlite db < file`) runs in the current shell and is fine.
+  #
+  # Written as a scan rather than as one case per site because the sites move:
+  # three were added the day this was found, by a change that was reviewed and
+  # cleared without anyone noticing the shell rule underneath it. A statement
+  # inside a helper that warms is reached through the helper, so the scan looks
+  # two lines up rather than one.
+  local root unwarmed
+  root="$(cd "$BATS_TEST_DIRNAME/.." && pwd)"
+  unwarmed="$(_462_unwarmed_under "$root")"
+  [ -z "$unwarmed" ] || {
+    printf 'piped agmsg_sqlite with no warm above it:\n%s\n' "$unwarmed"
+    false
+  }
+  # The scan can see something: a positive control on the instrument itself,
+  # counting only executable sites (a comment mentioning the pipe is not one).
+  [ "$(_462_piped_sites_under "$root")" -ge 8 ]
+}
+
+@test "storage: the #462 scan goes red when a helper loses its warm (mutation control)" {
+  # A guard that stays green under the break it exists for is not a guard.
+  # Copy the tree, delete the warm from the stdin helper -- the central path
+  # every driver batch goes through -- and the scan must name that pipe.
+  local root copy target
+  root="$(cd "$BATS_TEST_DIRNAME/.." && pwd)"
+  copy="$BATS_TEST_TMPDIR/scan-mutant"
+  mkdir -p "$copy"
+  cp -R "$root/scripts" "$copy/scripts"
+  target="$copy/scripts/drivers/storage/sqlite.sh"
+  grep -q 'agmsg_sqlite_warm' "$target"
+  # Remove exactly the warm line inside _sqlite_exec_stdin.
+  awk 'BEGIN{inside=0} /^_sqlite_exec_stdin\(\) \{/{inside=1} inside && /agmsg_sqlite_warm/{inside=0; next} /^}/{inside=0} {print}' "$target" > "$target.mutant"
+  mv "$target.mutant" "$target"
+  # The mutation took: one fewer warm in the file.
+  [ "$(grep -c 'agmsg_sqlite_warm' "$target")" -eq "$(( $(grep -c 'agmsg_sqlite_warm' "$root/scripts/drivers/storage/sqlite.sh") - 1 ))" ]
+  local unwarmed
+  unwarmed="$(_462_unwarmed_under "$copy")"
+  printf '%s\n' "$unwarmed" | grep -q '^scripts/drivers/storage/sqlite.sh:[0-9][0-9]*$'
+  # And ONLY that site: the mutant differs from the tree in one place.
+  [ "$(printf '%s\n' "$unwarmed" | grep -c .)" -eq 1 ]
+}