packmate/stress/network/run.sh

#!/bin/sh
set -eu

SCRIPT_DIR=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd)
REPO_DIR=$(CDPATH= cd -- "$SCRIPT_DIR/../.." && pwd)
COMPOSE_FILE="$SCRIPT_DIR/docker-compose.yml"
PROJECT_RAW=${PACKMATE_NETWORK_STRESS_PROJECT:-"packmate-network-stress-${CI_JOB_ID:-local}"}
PROJECT=$(printf '%s' "$PROJECT_RAW" | tr -cd 'a-zA-Z0-9_-' | tr 'A-Z' 'a-z')
RUN_LABEL_RAW=${PACKMATE_NETWORK_STRESS_RUN_ID:-"$(date -u +%Y%m%dT%H%M%SZ)-$$"}
RUN_LABEL=$(printf '%s' "$RUN_LABEL_RAW" | tr -cd 'a-zA-Z0-9_.-')
if [ -z "$PROJECT" ] || [ -z "$RUN_LABEL" ]; then
  echo "stress project and run ID must contain letters, digits, dot, underscore or dash" >&2
  exit 2
fi
REPORT_DIR="$REPO_DIR/build/reports/network-stress/$RUN_LABEL"
REPORT_DIR_DOCKER="$REPORT_DIR"
if [ -n "${MSYSTEM:-}" ]; then
  COMPOSE_FILE=$(cygpath -w "$COMPOSE_FILE")
  REPORT_DIR_DOCKER=$(cygpath -w "$REPORT_DIR")
fi

BASELINE_STREAMS=${PACKMATE_NETWORK_STRESS_STREAMS:-250000}
LIVE_STREAMS=${PACKMATE_NETWORK_STRESS_LIVE_STREAMS:-50000}
SHARDS=${PACKMATE_NETWORK_STRESS_SHARDS:-16}
CONCURRENCY=${PACKMATE_NETWORK_STRESS_CONCURRENCY:-128}
FRAMES=${PACKMATE_NETWORK_STRESS_FRAMES_PER_STREAM:-8}
PAYLOAD_BYTES=${PACKMATE_NETWORK_STRESS_PAYLOAD_BYTES:-1024}
DRAIN_TIMEOUT=${PACKMATE_NETWORK_STRESS_DRAIN_TIMEOUT_SECONDS:-1800}
CLEANUP_TIMEOUT=${PACKMATE_NETWORK_STRESS_CLEANUP_TIMEOUT_SECONDS:-1800}
MAX_API_P99_MS=${PACKMATE_NETWORK_STRESS_MAX_API_P99_MS:-500}
MAX_CAPTURE_QUEUE=${PACKMATE_NETWORK_STRESS_MAX_CAPTURE_QUEUE:-500000}
CAPTURE_WORKERS=${PACKMATE_NETWORK_STRESS_CAPTURE_WORKERS:-8}
CAPTURE_QUEUE_CAPACITY=${PACKMATE_NETWORK_STRESS_CAPTURE_QUEUE_CAPACITY:-250000}
CAPTURE_BUFFER_SIZE_BYTES=${PACKMATE_NETWORK_STRESS_CAPTURE_BUFFER_SIZE_BYTES:-67108864}
TCP_TIMEOUT_SECONDS=${PACKMATE_NETWORK_STRESS_TCP_TIMEOUT_SECONDS:-40}
UDP_TIMEOUT_SECONDS=${PACKMATE_NETWORK_STRESS_UDP_TIMEOUT_SECONDS:-20}
MIN_STREAMS_PER_SECOND=${PACKMATE_NETWORK_STRESS_MIN_STREAMS_PER_SECOND:-500}
MIN_CLEANUP_STREAMS_PER_SECOND=${PACKMATE_NETWORK_STRESS_MIN_CLEANUP_STREAMS_PER_SECOND:-$MIN_STREAMS_PER_SECOND}
MAX_INGEST_DRAIN_SECONDS=${PACKMATE_NETWORK_STRESS_MAX_INGEST_DRAIN_SECONDS:-120}
MIN_LIVE_DURATION_SECONDS=${PACKMATE_NETWORK_STRESS_MIN_LIVE_DURATION_SECONDS:-20}
FAVORITE_STREAMS=${PACKMATE_NETWORK_STRESS_FAVORITE_STREAMS:-100}
STORAGE_MULTIPLIER=${PACKMATE_NETWORK_STRESS_STORAGE_MULTIPLIER:-3}
TARGET_IP=${PACKMATE_NETWORK_STRESS_TARGET_IP:-172.31.250.10}
PACKMATE_NETWORK_STRESS_CLEANUP_INTERVAL_SECONDS=${PACKMATE_NETWORK_STRESS_CLEANUP_INTERVAL_SECONDS:-10}
PACKMATE_NETWORK_STRESS_CLEANUP_THRESHOLD_MINUTES=${PACKMATE_NETWORK_STRESS_CLEANUP_THRESHOLD_MINUTES:-1440}
PACKMATE_NETWORK_STRESS_CAPTURE_WORKERS=$CAPTURE_WORKERS
PACKMATE_NETWORK_STRESS_CAPTURE_QUEUE_CAPACITY=$CAPTURE_QUEUE_CAPACITY
PACKMATE_NETWORK_STRESS_CAPTURE_BUFFER_SIZE_BYTES=$CAPTURE_BUFFER_SIZE_BYTES
PACKMATE_NETWORK_STRESS_TCP_TIMEOUT_SECONDS=$TCP_TIMEOUT_SECONDS
PACKMATE_NETWORK_STRESS_UDP_TIMEOUT_SECONDS=$UDP_TIMEOUT_SECONDS
export PACKMATE_NETWORK_STRESS_CLEANUP_INTERVAL_SECONDS PACKMATE_NETWORK_STRESS_CLEANUP_THRESHOLD_MINUTES
export PACKMATE_NETWORK_STRESS_CAPTURE_WORKERS PACKMATE_NETWORK_STRESS_CAPTURE_QUEUE_CAPACITY
export PACKMATE_NETWORK_STRESS_CAPTURE_BUFFER_SIZE_BYTES
export PACKMATE_NETWORK_STRESS_TCP_TIMEOUT_SECONDS PACKMATE_NETWORK_STRESS_UDP_TIMEOUT_SECONDS
PACKMATE_NETWORK_STRESS_CLEANUP_ENABLED=false
export PACKMATE_NETWORK_STRESS_CLEANUP_ENABLED

require_positive() {
  case "$1" in ''|*[!0-9]*|0) echo "$2 must be a positive integer" >&2; exit 2 ;; esac
}

for setting in \
  "$BASELINE_STREAMS:PACKMATE_NETWORK_STRESS_STREAMS" \
  "$LIVE_STREAMS:PACKMATE_NETWORK_STRESS_LIVE_STREAMS" \
  "$SHARDS:PACKMATE_NETWORK_STRESS_SHARDS" \
  "$CONCURRENCY:PACKMATE_NETWORK_STRESS_CONCURRENCY" \
  "$FRAMES:PACKMATE_NETWORK_STRESS_FRAMES_PER_STREAM" \
  "$PAYLOAD_BYTES:PACKMATE_NETWORK_STRESS_PAYLOAD_BYTES" \
  "$MAX_CAPTURE_QUEUE:PACKMATE_NETWORK_STRESS_MAX_CAPTURE_QUEUE" \
  "$CAPTURE_WORKERS:PACKMATE_NETWORK_STRESS_CAPTURE_WORKERS" \
  "$CAPTURE_QUEUE_CAPACITY:PACKMATE_NETWORK_STRESS_CAPTURE_QUEUE_CAPACITY" \
  "$CAPTURE_BUFFER_SIZE_BYTES:PACKMATE_NETWORK_STRESS_CAPTURE_BUFFER_SIZE_BYTES" \
  "$TCP_TIMEOUT_SECONDS:PACKMATE_NETWORK_STRESS_TCP_TIMEOUT_SECONDS" \
  "$UDP_TIMEOUT_SECONDS:PACKMATE_NETWORK_STRESS_UDP_TIMEOUT_SECONDS" \
  "$MIN_STREAMS_PER_SECOND:PACKMATE_NETWORK_STRESS_MIN_STREAMS_PER_SECOND" \
  "$MIN_CLEANUP_STREAMS_PER_SECOND:PACKMATE_NETWORK_STRESS_MIN_CLEANUP_STREAMS_PER_SECOND" \
  "$MAX_INGEST_DRAIN_SECONDS:PACKMATE_NETWORK_STRESS_MAX_INGEST_DRAIN_SECONDS" \
  "$MIN_LIVE_DURATION_SECONDS:PACKMATE_NETWORK_STRESS_MIN_LIVE_DURATION_SECONDS" \
  "$FAVORITE_STREAMS:PACKMATE_NETWORK_STRESS_FAVORITE_STREAMS" \
  "$STORAGE_MULTIPLIER:PACKMATE_NETWORK_STRESS_STORAGE_MULTIPLIER" \
  "$PACKMATE_NETWORK_STRESS_CLEANUP_INTERVAL_SECONDS:PACKMATE_NETWORK_STRESS_CLEANUP_INTERVAL_SECONDS" \
  "$PACKMATE_NETWORK_STRESS_CLEANUP_THRESHOLD_MINUTES:PACKMATE_NETWORK_STRESS_CLEANUP_THRESHOLD_MINUTES"
do
  require_positive "${setting%%:*}" "${setting#*:}"
done
if [ "$PAYLOAD_BYTES" -lt 32 ]; then
  echo "PACKMATE_NETWORK_STRESS_PAYLOAD_BYTES must be at least 32" >&2
  exit 2
fi
if [ "$FAVORITE_STREAMS" -gt "$BASELINE_STREAMS" ]; then
  echo "favorite stream count cannot exceed baseline streams" >&2
  exit 2
fi

mkdir -p "$REPORT_DIR"

compose() {
  MSYS_NO_PATHCONV=1 docker compose --profile tools -p "$PROJECT" -f "$COMPOSE_FILE" "$@"
}

db_scalar() {
  output=$(compose exec -T db psql -v ON_ERROR_STOP=1 -U packmate -d packmate -Atqc "$1") || return $?
  printf '%s' "$output" | tr -d '\r'
}

db_integer() {
  value=$(db_scalar "$1") || return $?
  case "$value" in
    ''|*[!0-9]*) echo "database query did not return an unsigned integer: $value" >&2; return 1 ;;
  esac
  printf '%s' "$value"
}

save_logs() {
  compose logs --no-color > "$REPORT_DIR/compose.log" 2>&1 || true
}

copy_reports() {
  if ! reports_container=$(compose ps -aq reports 2>/dev/null); then
    echo "could not locate the stress reports container; preserving Docker volumes for recovery" >&2
    return 1
  fi
  if [ -z "$reports_container" ]; then
    echo "stress reports container does not exist; preserving Docker volumes for recovery" >&2
    return 1
  fi
  if ! MSYS_NO_PATHCONV=1 docker cp "$reports_container:/reports/." "$REPORT_DIR_DOCKER"; then
    echo "could not copy stress reports; preserving Docker volumes for recovery" >&2
    return 1
  fi
}

cleanup() {
  status=$1
  trap - 0 2 15
  reports_copied=true
  if ! copy_reports; then
    reports_copied=false
    status=1
  fi
  save_logs
  if [ "$reports_copied" = true ]; then
    compose down --volumes --remove-orphans >/dev/null 2>&1 || true
  else
    compose down --remove-orphans >/dev/null 2>&1 || true
  fi
  exit "$status"
}
trap 'cleanup $?' 0
trap 'cleanup 130' 2
trap 'cleanup 143' 15

wait_for_streams() {
  condition=$1
  minimum=$2
  timeout=$3
  label=$4
  watch_pid=${5:-}
  deadline=$(( $(date +%s) + timeout ))
  while :; do
    if [ -n "$watch_pid" ] && ! kill -0 "$watch_pid" 2>/dev/null; then
      echo "$label stopped because its health probe exited early" >&2
      return 1
    fi
    current=$(db_integer "SELECT count(*) FROM stream WHERE $condition")
    if [ "$current" -ge "$minimum" ]; then
      printf '%s reached %s streams (required %s)\n' "$label" "$current" "$minimum"
      return 0
    fi
    if [ "$(date +%s)" -ge "$deadline" ]; then
      echo "$label reached only $current of required $minimum streams" >&2
      return 1
    fi
    sleep 2
  done
}

wait_for_zero() {
  condition=$1
  timeout=$2
  label=$3
  watch_pid=${4:-}
  deadline=$(( $(date +%s) + timeout ))
  while :; do
    if [ -n "$watch_pid" ] && ! kill -0 "$watch_pid" 2>/dev/null; then
      echo "$label stopped because its health probe exited early" >&2
      return 1
    fi
    current=$(db_integer "SELECT count(*) FROM stream WHERE $condition")
    if [ "$current" -eq 0 ]; then
      printf '%s drained\n' "$label"
      return 0
    fi
    if [ "$(date +%s)" -ge "$deadline" ]; then
      echo "$label still has $current streams" >&2
      return 1
    fi
    sleep 2
  done
}

check_rate() {
  phase=$1
  streams=$2
  elapsed=$3
  if [ "$elapsed" -lt 1 ]; then elapsed=1; fi
  if [ "$streams" -lt $((MIN_STREAMS_PER_SECOND * elapsed)) ]; then
    echo "$phase achieved less than $MIN_STREAMS_PER_SECOND completed TCP streams/s ($streams in ${elapsed}s)" >&2
    return 1
  fi
  printf '%s completed %s streams in %ss (at least %s streams/s)\n' \
    "$phase" "$streams" "$elapsed" "$MIN_STREAMS_PER_SECOND"
}

load_phase_elapsed() {
  phase=$1
  compose exec -T reports python -c '
import glob, json, math, sys
paths = glob.glob(f"/reports/{sys.argv[1]}-*.json")
if not paths:
    raise SystemExit(f"no generator reports found for phase {sys.argv[1]}")
reports = [json.load(open(path, encoding="utf-8")) for path in paths]
reports = [report for report in reports if "startedAtUnixSeconds" in report]
if not reports:
    raise SystemExit(f"no timed generator reports found for phase {sys.argv[1]}")
elapsed = max(report["finishedAtUnixSeconds"] for report in reports) \
    - min(report["startedAtUnixSeconds"] for report in reports)
print(max(1, math.ceil(elapsed)))
' "$phase" | tr -d '\r'
}

emit_ledger() {
  stream_condition=$1
  compose exec -T db psql -v ON_ERROR_STOP=1 -U packmate -d packmate -At \
    -F "$(printf '\t')" -c "
      WITH payloads AS (
        SELECT s.id AS stream_id,
               string_agg(p.content, ''::bytea ORDER BY p.id) FILTER (WHERE p.incoming) AS incoming_content,
               string_agg(p.content, ''::bytea ORDER BY p.id) FILTER (WHERE NOT p.incoming) AS outgoing_content,
               bool_and(COALESCE(
                 CASE WHEN p.incoming
                   THEN p.destination_ip = '$TARGET_IP' AND p.source_ip <> '$TARGET_IP'
                   ELSE p.source_ip = '$TARGET_IP' AND p.destination_ip <> '$TARGET_IP'
                 END,
                 false
               )) AS ips_valid,
               count(p.id) = s.packets_count
                 AND sum(octet_length(p.content)) = s.size_bytes
                 AND s.service_id = 18080
                 AND s.protocol = 0
                 AND (s.end_timestamp = 0 OR s.start_timestamp <= s.end_timestamp)
                 AS metadata_valid
        FROM stream s
        JOIN packet p ON p.stream_id = s.id
        WHERE $stream_condition
        GROUP BY s.id, s.packets_count, s.size_bytes, s.service_id, s.protocol,
                 s.start_timestamp, s.end_timestamp
      )
      SELECT stream_id,
             encode(substring(incoming_content FROM 5 FOR 24), 'hex'),
             octet_length(incoming_content),
             md5(incoming_content),
             octet_length(outgoing_content),
             md5(outgoing_content),
             ips_valid,
             metadata_valid
      FROM payloads
      ORDER BY stream_id"
}

count_invalid_matches() {
  stream_condition=$1
  db_integer "
    SELECT count(*)
    FROM (
      SELECT s.id
      FROM stream s
      JOIN packet p ON p.stream_id = s.id
      LEFT JOIN found_pattern fp ON fp.packet_id = p.id
      WHERE $stream_condition
      GROUP BY s.id
      HAVING count(fp.id) <> 2
         OR count(fp.id) FILTER (WHERE p.incoming) <> 1
         OR count(fp.id) FILTER (WHERE NOT p.incoming) <> 1
         OR bool_and(
              substring(
                p.content
                FROM fp.start_position + 1
                FOR fp.end_position - fp.start_position + 1
              ) = convert_to('PMSTRESS', 'UTF8')
              AND COALESCE(fp.search_target, 'CONTENT') = 'CONTENT'
              AND fp.pattern_id = (SELECT id FROM pattern WHERE name = 'network-stress-marker')
            ) FILTER (WHERE fp.id IS NOT NULL) IS NOT TRUE
    ) invalid"
}

verify_complete_ledger() {
  condition=$1
  run_id=$2
  total=$3
  expected=$4
  report=$5
  subset=${6:-false}
  subset_arg=""
  ledger_file="$REPORT_DIR/$report.tsv"
  if [ "$subset" = true ]; then subset_arg="--allow-subset"; fi
  emit_ledger "$condition" > "$ledger_file"
  # shellcheck disable=SC2086
  compose run --rm -T --no-deps generator \
    python /stress/verify_ledger.py \
      --run-id "$run_id" \
      --total-streams "$total" \
      --expected-streams "$expected" \
      --shards "$SHARDS" \
      --frames "$FRAMES" \
      --payload-bytes "$PAYLOAD_BYTES" \
      --report "/reports/$report" \
      $subset_arg < "$ledger_file"
  rm -f "$ledger_file"
}

run_load_phase() {
  phase=$1
  total=$2
  run_id=$3
  abort_file=$4
  minimum_duration=$5
  base=$((total / SHARDS))
  remainder=$((total % SHARDS))
  pids=""
  shard=0
  offset=0
  while [ "$shard" -lt "$SHARDS" ]; do
    count=$base
    if [ "$shard" -lt "$remainder" ]; then count=$((count + 1)); fi
    if [ "$count" -gt 0 ]; then
      compose run --rm --no-deps generator python /stress/network_load.py \
        --streams "$count" \
        --frames "$FRAMES" \
        --payload-bytes "$PAYLOAD_BYTES" \
        --concurrency "$CONCURRENCY" \
        --run-id "$run_id" \
        --shard "$shard" \
        --stream-offset "$offset" \
        --abort-file "$abort_file" \
        --minimum-duration-seconds "$minimum_duration" \
        --report "/reports/$phase-$shard.json" &
      pids="$pids $!"
    fi
    offset=$((offset + count))
    shard=$((shard + 1))
  done
  failed=0
  for pid in $pids; do
    if ! wait "$pid"; then failed=1; fi
  done
  return "$failed"
}

echo "Starting isolated network stress project $PROJECT; reports: $REPORT_DIR"
compose up -d --build db target packmate reports
compose build generator
compose exec -T reports sh -c 'rm -f /reports/*'
compose run --rm --no-deps generator python /stress/api_control.py prepare --timeout 300

db_df=$(compose exec -T db df -PkT /var/lib/postgresql/data) || exit $?
db_available_kb=$(printf '%s\n' "$db_df" | awk 'NR == 2 {print $5}' | tr -d '\r')
db_storage_type=$(printf '%s\n' "$db_df" | awk 'NR == 2 {print $2}' | tr -d '\r')
require_positive "$db_available_kb" "database available storage"
estimated_raw_bytes=$(( (BASELINE_STREAMS + LIVE_STREAMS) * FRAMES * (PAYLOAD_BYTES + 4) * 2 ))
required_storage_bytes=$((estimated_raw_bytes * STORAGE_MULTIPLIER))
available_storage_bytes=$((db_available_kb * 1024))
if [ "$available_storage_bytes" -lt "$required_storage_bytes" ]; then
  echo "database volume has $available_storage_bytes bytes free; stress preflight requires $required_storage_bytes" >&2
  exit 1
fi

cat > "$REPORT_DIR/effective-config.json" <<EOF
{
  "project": "$PROJECT",
  "runId": "$RUN_LABEL",
  "baselineStreams": $BASELINE_STREAMS,
  "concurrentStreams": $LIVE_STREAMS,
  "shards": $SHARDS,
  "concurrencyPerShard": $CONCURRENCY,
  "framesPerStream": $FRAMES,
  "payloadBytes": $PAYLOAD_BYTES,
  "minimumLiveDurationSeconds": $MIN_LIVE_DURATION_SECONDS,
  "maximumCaptureQueue": $MAX_CAPTURE_QUEUE,
  "captureWorkers": $CAPTURE_WORKERS,
  "captureQueueCapacity": $CAPTURE_QUEUE_CAPACITY,
  "captureBufferSizeBytes": $CAPTURE_BUFFER_SIZE_BYTES,
  "tcpTimeoutSeconds": $TCP_TIMEOUT_SECONDS,
  "udpTimeoutSeconds": $UDP_TIMEOUT_SECONDS,
  "minimumStreamsPerSecond": $MIN_STREAMS_PER_SECOND,
  "minimumCleanupStreamsPerSecond": $MIN_CLEANUP_STREAMS_PER_SECOND,
  "cleanupIntervalSeconds": $PACKMATE_NETWORK_STRESS_CLEANUP_INTERVAL_SECONDS,
  "cleanupRetentionMinutes": $PACKMATE_NETWORK_STRESS_CLEANUP_THRESHOLD_MINUTES,
  "targetIp": "$TARGET_IP",
  "databaseVolumeFreeBytesAtStart": $available_storage_bytes,
  "databaseStorageType": "$db_storage_type"
}
EOF

baseline_started=$(date +%s)
compose run --rm --no-deps generator python /stress/api_control.py monitor \
  --stop-file /reports/baseline-load.done \
  --abort-file /reports/baseline-load.abort \
  --max-queue "$MAX_CAPTURE_QUEUE" \
  --report /reports/baseline-capture-monitor.json &
baseline_monitor_pid=$!
baseline_load_status=0
if ! run_load_phase baseline "$BASELINE_STREAMS" 1 /reports/baseline-load.abort 0; then
  baseline_load_status=1
fi
compose exec -T reports touch /reports/baseline-load.done
baseline_monitor_status=0
if ! wait "$baseline_monitor_pid"; then baseline_monitor_status=1; fi
if [ "$baseline_load_status" -ne 0 ] || [ "$baseline_monitor_status" -ne 0 ]; then
  echo "baseline traffic or continuous capture monitor failed" >&2
  exit 1
fi
baseline_load_elapsed=$(load_phase_elapsed baseline)
require_positive "$baseline_load_elapsed" "baseline generator elapsed time"
check_rate "baseline generators" "$BASELINE_STREAMS" "$baseline_load_elapsed"
compose run --rm --no-deps generator python /stress/api_control.py metrics \
  --max-queue "$MAX_CAPTURE_QUEUE" \
  --report /reports/pcap-metrics-after-baseline-load.json
baseline_drain_started=$(date +%s)
wait_for_streams "end_timestamp > 0 AND favorite = false" "$BASELINE_STREAMS" "$DRAIN_TIMEOUT" "baseline network ingest"
baseline_drain_seconds=$(( $(date +%s) - baseline_drain_started ))
if [ "$baseline_drain_seconds" -gt "$MAX_INGEST_DRAIN_SECONDS" ]; then
  echo "baseline ingest queue needed ${baseline_drain_seconds}s to drain" >&2
  exit 1
fi

compose run --rm --no-deps generator python /stress/api_control.py metrics \
  --require-empty-queue \
  --wait-seconds "$DRAIN_TIMEOUT" \
  --report /reports/pcap-metrics-before-cleanup.json

compose run --rm --no-deps generator python /stress/api_control.py verify \
  --run-id 1 \
  --sample-size 200 \
  --frames "$FRAMES" \
  --payload-bytes "$PAYLOAD_BYTES" \
  --report /reports/baseline-payload-verification.json

baseline_captured=$(db_integer "SELECT count(*) FROM stream")
baseline_packets=$(db_integer "SELECT count(*) FROM packet")
baseline_matches=$(db_integer "SELECT count(*) FROM found_pattern")
baseline_pattern_links=$(db_integer "SELECT count(*) FROM stream_found_patterns")
invalid_baseline_matches=$(count_invalid_matches "true")
if [ "$baseline_captured" -ne "$BASELINE_STREAMS" ] \
    || [ "$baseline_matches" -ne $((BASELINE_STREAMS * 2)) ] \
    || [ "$baseline_pattern_links" -ne "$BASELINE_STREAMS" ] \
    || [ "$invalid_baseline_matches" -ne 0 ]; then
  echo "invalid baseline counts: streams=$baseline_captured matches=$baseline_matches links=$baseline_pattern_links invalidMatches=$invalid_baseline_matches" >&2
  exit 1
fi

verify_complete_ledger "true" 1 "$BASELINE_STREAMS" "$BASELINE_STREAMS" baseline-full-ledger.json

db_scalar "UPDATE stream SET favorite = true WHERE id IN (SELECT id FROM stream ORDER BY id LIMIT $FAVORITE_STREAMS)" >/dev/null
favorite_count=$(db_integer "SELECT count(*) FROM stream WHERE favorite")
if [ "$favorite_count" -ne "$FAVORITE_STREAMS" ]; then
  echo "could not mark $FAVORITE_STREAMS baseline streams as favorite" >&2
  exit 1
fi

db_bytes_before=$(db_integer "SELECT pg_database_size(current_database())")
wal_start=$(db_scalar "SELECT pg_current_wal_lsn()")
db_scalar "UPDATE stream SET end_timestamp = 0, persisted_at = 0" >/dev/null

eligible_baseline=$(db_integer "SELECT count(*) FROM stream WHERE end_timestamp = 0 AND favorite = false")
expected_eligible=$((BASELINE_STREAMS - FAVORITE_STREAMS))
if [ "$eligible_baseline" -ne "$expected_eligible" ]; then
  echo "could not create the complete eligible baseline: eligible=$eligible_baseline expected=$expected_eligible" >&2
  exit 1
fi
cleanup_measurement_started_at=$(date +%s)
printf 'unix_seconds,eligible_baseline_streams,committed_live_streams\n' > "$REPORT_DIR/cleanup-overlap.csv"
printf '%s,%s,0\n' "$cleanup_measurement_started_at" "$eligible_baseline" >> "$REPORT_DIR/cleanup-overlap.csv"

PACKMATE_NETWORK_STRESS_CLEANUP_ENABLED=true
export PACKMATE_NETWORK_STRESS_CLEANUP_ENABLED
compose up -d --no-deps --force-recreate packmate
compose run --rm --no-deps generator python /stress/api_control.py prepare --timeout 300

compose run --rm --no-deps generator python /stress/api_control.py probe \
  --requests 20000 \
  --pause 0.1 \
  --min-requests 100 \
  --stop-file /reports/workload.done \
  --abort-file /reports/live-load.abort \
  --max-queue "$MAX_CAPTURE_QUEUE" \
  --max-p99-ms "$MAX_API_P99_MS" \
  --report /reports/api-probe.json &
probe_pid=$!
run_load_phase live "$LIVE_STREAMS" 2 /reports/live-load.abort "$MIN_LIVE_DURATION_SECONDS" &
live_pid=$!

overlap_observed=false
cleanup_completed_at=0
previous_old=$eligible_baseline
previous_live=0
while kill -0 "$live_pid" 2>/dev/null; do
  sample_time=$(date +%s)
  remaining_old=$(db_integer "SELECT count(*) FROM stream WHERE end_timestamp = 0 AND favorite = false")
  committed_live=$(db_integer "SELECT count(*) FROM stream WHERE end_timestamp > 0 AND favorite = false")
  printf '%s,%s,%s\n' "$sample_time" "$remaining_old" "$committed_live" >> "$REPORT_DIR/cleanup-overlap.csv"
  if [ "$remaining_old" -lt "$previous_old" ] \
      && [ "$committed_live" -gt "$previous_live" ] \
      && [ "$previous_live" -gt 0 ]; then
    overlap_observed=true
  fi
  if [ "$remaining_old" -eq 0 ] && [ "$cleanup_completed_at" -eq 0 ]; then
    cleanup_completed_at=$sample_time
  fi
  if ! kill -0 "$probe_pid" 2>/dev/null; then
    echo "API/capture probe exited while concurrent traffic was active" >&2
    compose exec -T reports touch /reports/live-load.abort
  fi
  previous_old=$remaining_old
  previous_live=$committed_live
  sleep 1
done

live_load_status=0
if ! wait "$live_pid"; then live_load_status=1; fi
if [ "$live_load_status" -ne 0 ]; then
  compose exec -T reports touch /reports/workload.done
  wait "$probe_pid" || true
  echo "concurrent traffic was aborted or a generator failed" >&2
  exit 1
fi
if [ "$overlap_observed" != true ]; then
  compose exec -T reports touch /reports/workload.done
  wait "$probe_pid" || true
  echo "cleanup did not overlap committed live network traffic" >&2
  exit 1
fi
live_load_elapsed=$(load_phase_elapsed live)
require_positive "$live_load_elapsed" "live generator elapsed time"
check_rate "live generators" "$LIVE_STREAMS" "$live_load_elapsed"
live_drain_started=$(date +%s)
wait_for_streams "end_timestamp > 0 AND favorite = false" "$LIVE_STREAMS" "$DRAIN_TIMEOUT" "concurrent network ingest" "$probe_pid"
live_drain_seconds=$(( $(date +%s) - live_drain_started ))
if [ "$live_drain_seconds" -gt "$MAX_INGEST_DRAIN_SECONDS" ]; then
  echo "live ingest queue needed ${live_drain_seconds}s to drain" >&2
  exit 1
fi
wait_for_zero "end_timestamp = 0 AND favorite = false" "$CLEANUP_TIMEOUT" "eligible cleanup backlog" "$probe_pid"
if [ "$cleanup_completed_at" -eq 0 ]; then cleanup_completed_at=$(date +%s); fi
cleanup_elapsed_seconds=$((cleanup_completed_at - cleanup_measurement_started_at))
if [ "$cleanup_elapsed_seconds" -lt 1 ]; then cleanup_elapsed_seconds=1; fi
cleanup_streams_per_second=$((eligible_baseline / cleanup_elapsed_seconds))
if [ "$cleanup_streams_per_second" -lt "$MIN_CLEANUP_STREAMS_PER_SECOND" ]; then
  echo "cleanup sustained only $cleanup_streams_per_second streams/s; required $MIN_CLEANUP_STREAMS_PER_SECOND" >&2
  exit 1
fi
compose run --rm -T --no-deps generator sh -c 'touch /reports/workload.done'
wait "$probe_pid"

compose run --rm --no-deps generator python /stress/api_control.py verify \
  --run-id 2 \
  --sample-size 200 \
  --frames "$FRAMES" \
  --payload-bytes "$PAYLOAD_BYTES" \
  --report /reports/concurrent-payload-verification.json

verify_complete_ledger "s.favorite = false" 2 "$LIVE_STREAMS" "$LIVE_STREAMS" concurrent-full-ledger.json
verify_complete_ledger "s.favorite = true" 1 "$BASELINE_STREAMS" "$FAVORITE_STREAMS" favorite-full-ledger.json true

compose run --rm --no-deps generator python /stress/api_control.py metrics \
  --require-empty-queue \
  --wait-seconds "$DRAIN_TIMEOUT" \
  --report /reports/pcap-metrics.json

orphan_packets=$(db_integer "SELECT count(*) FROM packet p LEFT JOIN stream s ON s.id=p.stream_id WHERE s.id IS NULL")
orphan_matches=$(db_integer "SELECT count(*) FROM found_pattern fp LEFT JOIN packet p ON p.id=fp.packet_id WHERE p.id IS NULL")
orphan_pattern_links=$(db_integer "SELECT count(*) FROM stream_found_patterns sfp LEFT JOIN stream s ON s.id=sfp.stream_id LEFT JOIN pattern pt ON pt.id=sfp.pattern_id WHERE s.id IS NULL OR pt.id IS NULL")
cleanup_pending=$(db_integer "SELECT count(*) FROM stream WHERE cleanup_pending")
final_matches=$(db_integer "SELECT count(*) FROM found_pattern")
final_pattern_links=$(db_integer "SELECT count(*) FROM stream_found_patterns")
invalid_final_matches=$(count_invalid_matches "true")
if [ "$orphan_packets" -ne 0 ] \
    || [ "$orphan_matches" -ne 0 ] \
    || [ "$orphan_pattern_links" -ne 0 ] \
    || [ "$cleanup_pending" -ne 0 ] \
    || [ "$final_matches" -ne $(((LIVE_STREAMS + FAVORITE_STREAMS) * 2)) ] \
    || [ "$final_pattern_links" -ne $((LIVE_STREAMS + FAVORITE_STREAMS)) ] \
    || [ "$invalid_final_matches" -ne 0 ]; then
  echo "invalid final state: orphan_packets=$orphan_packets orphan_matches=$orphan_matches orphan_links=$orphan_pattern_links cleanup_pending=$cleanup_pending matches=$final_matches links=$final_pattern_links invalidMatches=$invalid_final_matches" >&2
  exit 1
fi

live_captured=$(db_integer "SELECT count(*) FROM stream WHERE end_timestamp > 0 AND favorite = false")
favorite_captured=$(db_integer "SELECT count(*) FROM stream WHERE favorite")
final_packets=$(db_integer "SELECT count(*) FROM packet")
db_bytes_after=$(db_integer "SELECT pg_database_size(current_database())")
wal_bytes=$(db_integer "SELECT pg_wal_lsn_diff(pg_current_wal_lsn(), '$wal_start')::bigint")
elapsed=$(( $(date +%s) - baseline_started ))

cat > "$REPORT_DIR/summary.json" <<EOF
{
  "baselineRequestedStreams": $BASELINE_STREAMS,
  "baselineCapturedStreams": $baseline_captured,
  "baselineCapturedPackets": $baseline_packets,
  "baselineMatches": $baseline_matches,
  "baselinePatternLinks": $baseline_pattern_links,
  "concurrentRequestedStreams": $LIVE_STREAMS,
  "concurrentCapturedStreams": $live_captured,
  "favoriteStreamsPreserved": $favorite_captured,
  "finalPackets": $final_packets,
  "finalMatches": $final_matches,
  "finalPatternLinks": $final_pattern_links,
  "shards": $SHARDS,
  "framesPerStream": $FRAMES,
  "payloadBytes": $PAYLOAD_BYTES,
  "captureWorkers": $CAPTURE_WORKERS,
  "captureQueueCapacity": $CAPTURE_QUEUE_CAPACITY,
  "captureBufferSizeBytes": $CAPTURE_BUFFER_SIZE_BYTES,
  "tcpTimeoutSeconds": $TCP_TIMEOUT_SECONDS,
  "elapsedSeconds": $elapsed,
  "baselineLoadSeconds": $baseline_load_elapsed,
  "baselineIngestDrainSeconds": $baseline_drain_seconds,
  "liveLoadSeconds": $live_load_elapsed,
  "liveIngestDrainSeconds": $live_drain_seconds,
  "minimumStreamsPerSecond": $MIN_STREAMS_PER_SECOND,
  "cleanupOverlapObserved": $overlap_observed,
  "cleanupElapsedSeconds": $cleanup_elapsed_seconds,
  "cleanupStreamsPerSecond": $cleanup_streams_per_second,
  "minimumCleanupStreamsPerSecond": $MIN_CLEANUP_STREAMS_PER_SECOND,
  "walBytesDuringCleanup": $wal_bytes,
  "databaseBytesBeforeCleanup": $db_bytes_before,
  "databaseBytesAfterCleanup": $db_bytes_after,
  "databaseVolumeFreeBytesAtStart": $available_storage_bytes,
  "databaseStorageType": "$db_storage_type",
  "estimatedRawTrafficBytes": $estimated_raw_bytes,
  "requiredStoragePreflightBytes": $required_storage_bytes,
  "orphanPackets": $orphan_packets,
  "orphanMatches": $orphan_matches,
  "orphanPatternLinks": $orphan_pattern_links,
  "cleanupPending": $cleanup_pending
}
EOF

cat "$REPORT_DIR/summary.json"
echo "Network capture and cleanup stress PASSED"