fix(bin): distinguish remote read deadlines from local ghost bounds

bin/fm-procevent-remote-reply.sh

@@ -19,8 +19,9 @@
 # ingests it, acknowledges the captured generation, then registers the next
 # cursor-anchored source. `relisten` tells that runner to poll again in the same
 # process, still holding the claim, after an empty window and after that re-arm.
-# A window the remote job worker preempted is reported to the runner as an empty
-# window, so it relistens too (see JOB_PREEMPTED below).
+# A window the remote job worker preempted, or one that hit its hard wall-clock
+# bound because the transport went ghost, is reported to the runner as an empty
+# window, so it relistens too (see JOB_PREEMPTED and BOUND_SECONDS below).
 # A continuity break is escalated and not re-armed, so the registration is dropped
 # and the runner stops. The runner does not refresh the owner lease.
 #
@@ -79,6 +80,16 @@ DATA="${FM_DATA_OVERRIDE:-$FM_HOME/data}"
 CURSOR_DIR="$STATE/remote-replies"
 REMOTE_LOG='state/parent-replies.status'
 WAIT_SECONDS=${FM_REMOTE_REPLY_WAIT_SECONDS:-55}
+# Hard wall-clock bound on one poll, window plus transport and queue slack.
+# The read itself ends after WAIT_SECONDS, so a poll still running at
+# BOUND_SECONDS is a ghost: the remote end died without closing the link,
+# which keepalive probes do not always convict. Killing it loses nothing -
+# the read is cursor-anchored and non-destructive - and the runner relistens
+# exactly as after a preemption, publishing no caught-up watermark.
+# A timeout status can also be the remote job worker's OWN execution deadline
+# reporting through fm-on.sh; cmd_source separates the two by elapsed time,
+# because only a remote deadline can arrive before the local bound fires.
+BOUND_SECONDS=${FM_REMOTE_REPLY_BOUND_SECONDS:-$((WAIT_SECONDS + 120))}
 MAX_DOC_BYTES=${FM_REMOTE_REPLY_MAX_DOC_BYTES:-262144}
 # fm-on.sh returns ssh's status unchanged, so 255 alone means unavailable
 # transport or unknown remote completion. Any other nonzero status is the remote
@@ -95,6 +106,8 @@ DOCUMENT_LOCAL_FAILURE=2
 . "$SCRIPT_DIR/fm-secondmate-registry-lib.sh"
 # shellcheck source=bin/fm-pending-reply-lib.sh
 . "$SCRIPT_DIR/fm-pending-reply-lib.sh"
+# shellcheck source=bin/fm-timeout-lib.sh
+. "$SCRIPT_DIR/fm-timeout-lib.sh"
 
 die() { printf 'error: %s\n' "$1" >&2; exit 1; }
 usage() { sed -n '2,66p' "$0" | sed 's/^# \{0,1\}//'; exit 2; }
@@ -268,16 +281,38 @@ WINDOW_CLOSED_EMPTY=75
 JOB_PREEMPTED=76
 
 cmd_source() {
-  local id=${1:-} started rc=0
+  local id=${1:-} started rc=0 elapsed
   validate_id "$id"
+  case "$BOUND_SECONDS" in
+    '' | *[!0-9]*)
+      die "FM_REMOTE_REPLY_BOUND_SECONDS must be a positive integer of seconds, got '$BOUND_SECONDS'"
+      ;;
+  esac
+  # All-digit strings can still be invalid octal literals (08, 09) or encode
+  # zero (00): force base 10 before any arithmetic, then reject non-positive.
+  BOUND_SECONDS=$((10#$BOUND_SECONDS))
+  if [ "$BOUND_SECONDS" -le 0 ]; then
+    die "FM_REMOTE_REPLY_BOUND_SECONDS must be a positive integer of seconds, got '$BOUND_SECONDS'"
+  fi
   read_cursor "$id"
   started=$(fm_pending_reply_now)
-  "$SCRIPT_DIR/fm-on.sh" "$id" fm-remote-delta-read.sh \
+  fm_run_timed "$BOUND_SECONDS" "$SCRIPT_DIR/fm-on.sh" "$id" fm-remote-delta-read.sh \
     "$REMOTE_LOG" "$CURSOR_OFFSET" "$CURSOR_HASH" "$WAIT_SECONDS" < /dev/null || rc=$?
   if [ "$rc" -eq "$WINDOW_CLOSED_EMPTY" ]; then
     fm_pending_reply_note_remote_channel_caught_up "$STATE" "$id" "$started" || true
   elif [ "$rc" -eq "$JOB_PREEMPTED" ]; then
     rc=$WINDOW_CLOSED_EMPTY
+  elif fm_timed_out "$rc"; then
+    # A timeout before the local bound could have fired is the remote job
+    # worker's own execution deadline (bin/fm-remote-job-lib.sh): the read
+    # FAILED, so exit like any other failed read instead of relistening into
+    # a worker that may keep missing its deadline. Epoch truncation can hide
+    # one second, so the guard band is one second under the bound.
+    elapsed=$(($(fm_pending_reply_now) - started))
+    if [ "$elapsed" -lt $((BOUND_SECONDS - 1)) ]; then
+      return "$rc"
+    fi
+    rc=$WINDOW_CLOSED_EMPTY
   fi
   return "$rc"
 }

tests/fm-remote-reply.test.sh

@@ -53,6 +53,10 @@ if [ -n "${FM_REMOTE_REPLY_POLL_LOG:-}" ]; then
   printf 'x\n' >> "$FM_REMOTE_REPLY_POLL_LOG"
 fi
 [ "${FM_REMOTE_REPLY_FAIL_READ:-}" != 1 ] || exit 255
+# Ghost transport: the remote end never answers and never closes the link.
+[ -z "${FM_REMOTE_REPLY_HANG_READ:-}" ] || { sleep 3600; exit 90; }
+# The remote job worker's own execution deadline, reporting through fm-on.
+[ -z "${FM_REMOTE_REPLY_REMOTE_TIMEOUT_READ:-}" ] || exit 124
 host=$1
 entry=$2
 shift 2
@@ -899,6 +903,45 @@ assert_absent "$PARENT/state/remote-replies/ios.caught-up" \
   "a preempted reply poll published a caught-up watermark"
 pass "a preempted reply poll reports a closed window without publishing channel freshness"
 
+# A ghost transport - the remote end died without closing the link - holds the
+# read open past its window, and keepalive probes do not always convict it.
+# The poll's hard bound kills it and reports a closed window so the runner
+# relistens; like a preemption it proves no caught-up watermark. An outer 20s
+# deadline keeps a bound regression from stalling the suite on the fixture's
+# hour-long sleep: a working bound exits 75 within about 3 seconds.
+rm -f -- "$PARENT/state/remote-replies/ios.caught-up"
+ghost_started=$(date +%s)
+set +e
+# shellcheck disable=SC2016 # Positional parameters expand in the child shell.
+remote_env env FM_REMOTE_REPLY_HANG_READ=1 FM_REMOTE_REPLY_WAIT_SECONDS=60 FM_REMOTE_REPLY_BOUND_SECONDS=3 \
+  bash -c '. "$1/bin/fm-timeout-lib.sh" || exit 97; fm_run_timed 20 "$2" source "$3"' \
+  _ "$ROOT" "$ADAPTER" ios > "$TMP_ROOT/ghost-source.out" 2>&1
+ghost_rc=$?
+set -e
+ghost_elapsed=$(($(date +%s) - ghost_started))
+[ "$ghost_rc" -eq 75 ] \
+  || fail "a ghost reply poll did not report a closed window inside its bound: $ghost_rc"
+[ "$ghost_elapsed" -lt 15 ] \
+  || fail "a ghost reply poll took ${ghost_elapsed}s, past its 3s bound and toward the outer 20s deadline"
+assert_absent "$PARENT/state/remote-replies/ios.caught-up" \
+  "a ghost reply poll published a caught-up watermark"
+pass "a ghost reply poll dies at its hard bound and relistens without a watermark"
+
+# The remote job worker's own execution deadline also surfaces as a timeout
+# status, but it means the read FAILED: the poll must exit like any other
+# failed read rather than relisten into a worker that may keep missing its
+# deadline, and it proves nothing about channel freshness.
+rm -f -- "$PARENT/state/remote-replies/ios.caught-up"
+set +e
+FM_REMOTE_REPLY_REMOTE_TIMEOUT_READ=1 remote_env "$ADAPTER" source ios > /dev/null 2>&1
+remote_deadline_rc=$?
+set -e
+[ "$remote_deadline_rc" -eq 124 ] \
+  || fail "a remote read deadline was relistened as a closed window: $remote_deadline_rc"
+assert_absent "$PARENT/state/remote-replies/ios.caught-up" \
+  "a remote read deadline published a caught-up watermark"
+pass "a remote read deadline exits as a failed read instead of relistening"
+
 # The per-cycle liveness probe is a non-preemptible job for the same remote home,
 # so the job worker preempts the listener's long-poll on every watcher cycle.
 # That must not cost the listener: it keeps its claim and polls again, and the