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

scripts/drivers/storage/sqlite-sync.sh

@@ -26,6 +26,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 "$db" >/dev/null 2>&1
 }
 
@@ -768,7 +769,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;
@@ -809,7 +810,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 -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'
@@ -1233,7 +1234,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
@@ -1328,7 +1329,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 -batch "$db" >/dev/null || return 13
+    COMMIT;" >/dev/null || 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

@@ -53,9 +53,18 @@ _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.
+_sqlite_exec_stdin() {
+  agmsg_sqlite_warm
+  printf '%s\n' "$2" | agmsg_sqlite -batch "$1"
+}
+
 # IN (...) list of "team:agent" pairs.
 _sqlite_pair_in() {
   local out="" p t a
@@ -252,8 +261,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"
@@ -497,6 +508,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

@@ -935,3 +935,40 @@ _longest_argv() {
   # cursor or a count moving by a few characters is not that.
   [ "$long" -lt "$((short + 200))" ]
 }
+
+@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.
+  local unwarmed="" file line n prev
+  while IFS=: read -r file n line; do
+    case "$file" in */lib/storage.sh) continue ;; esac
+    # The two lines above the pipeline: `agmsg_sqlite_warm`, or a helper whose
+    # own body warms.
+    prev="$(sed -n "$((n > 2 ? n - 2 : 1)),$((n))p" "$BATS_TEST_DIRNAME/../$file")"
+    case "$prev" in
+      *agmsg_sqlite_warm*) continue ;;
+    esac
+    unwarmed="$unwarmed$file:$n
+"
+  done < <(cd "$BATS_TEST_DIRNAME/.." && grep -rn '| agmsg_sqlite' scripts/)
+
+  # A statement inside a helper that warms is reached through the helper, so the
+  # scan looks two lines up rather than one.
+  [ -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.
+  local found
+  found="$(cd "$BATS_TEST_DIRNAME/.." && grep -rc '| agmsg_sqlite' scripts/ | awk -F: '{s+=$2} END {print s}')"
+  [ "$found" -ge 8 ]
+}