From de6eb084e5ad31cc10bb9d165da4022c5e3a7d94 Mon Sep 17 00:00:00 2001 From: weichen Date: Mon, 12 Oct 2026 14:20:13 +0000 Subject: [PATCH] Measure C4 recovery on fresh conversations Why: The first pg-stop window counted an in-flight 200 at shutdown, and five writes on the timeline conversation returned 409 while a mock run held it. That measured the probe, not a proven database failure. Evidence: go test ./... in tests/ops/drill_driver passed locally (SELF). Impact: The outage clock starts 2s after stop or kill. RTO is five writes on new conversations. Idempotency replays only the window and retries 409 for up to 2 minutes. The Redis phase no longer uses local outside a function. Refs: feature/u7-ops-drills Co-authored-by: Cursor --- tests/ops/drill_driver/main.go | 92 ++++++++++++++++++++++++--- tests/ops/drill_driver/window_test.go | 21 ++++++ tests/ops/run_fault_drill.sh | 36 +++++++---- 3 files changed, 128 insertions(+), 21 deletions(-) create mode 100644 tests/ops/drill_driver/window_test.go diff --git a/tests/ops/drill_driver/main.go b/tests/ops/drill_driver/main.go index 465fdd78..2271f599 100644 --- a/tests/ops/drill_driver/main.go +++ b/tests/ops/drill_driver/main.go @@ -238,6 +238,9 @@ func cmdIdem(args []string) error { conv := fs.String("conv", "", "conversation id") in := fs.String("in", "", "timeline json") out := fs.String("out", "", "json output") + since := fs.String("since", "", "only replay attempts at or after this time") + until := fs.String("until", "", "only replay attempts at or before this time") + retry := fs.Duration("retry", 2*time.Minute, "retry 409 until the conversation is free") if err := fs.Parse(args); err != nil { return err } @@ -255,22 +258,69 @@ func cmdIdem(args []string) error { if err != nil { return err } - var replays []attempt + var pending []attempt for _, a := range doc.Attempts { if a.Status >= 200 && a.Status < 300 { continue } - res, _ := s.do("POST", "/api/conversations/"+*conv+"/messages", map[string]any{ - "text": "drill " + a.RequestID, "requestId": a.RequestID, - }) - replays = append(replays, attempt{ - RequestID: a.RequestID, Base: *base, Started: time.Now().UTC().Format(time.RFC3339Nano), - ElapsedMs: res.ElapsedMs, Status: res.Status, Error: res.Error, MessageID: messageID(res.Body), - }) + if !attemptInWindow(a.Started, *since, *until) { + continue + } + pending = append(pending, a) + } + var replays []attempt + deadline := time.Now().Add(*retry) + for len(pending) > 0 { + var again []attempt + for _, a := range pending { + res, _ := s.do("POST", "/api/conversations/"+*conv+"/messages", map[string]any{ + "text": "drill " + a.RequestID, "requestId": a.RequestID, + }) + // 409 发生在写入之前：会话仍被 mock run 占用，requestId 还没落库。 + if res.Status == http.StatusConflict && time.Now().Before(deadline) { + again = append(again, a) + continue + } + replays = append(replays, attempt{ + RequestID: a.RequestID, Base: *base, Started: time.Now().UTC().Format(time.RFC3339Nano), + ElapsedMs: res.ElapsedMs, Status: res.Status, Error: res.Error, MessageID: messageID(res.Body), + }) + } + pending = again + if len(pending) > 0 { + time.Sleep(time.Second) + } } return writeJSON(*out, map[string]any{"replays": replays}) } +func parseStamp(s string) (time.Time, error) { + if t, err := time.Parse(time.RFC3339Nano, s); err == nil { + return t, nil + } + return time.Parse(time.RFC3339, s) +} + +func attemptInWindow(started, since, until string) bool { + t, err := parseStamp(started) + if err != nil { + return false + } + if since != "" { + s, err := parseStamp(since) + if err != nil || t.Before(s) { + return false + } + } + if until != "" { + u, err := parseStamp(until) + if err != nil || t.After(u) { + return false + } + } + return true +} + func cmdSSE(args []string) error { fs := flag.NewFlagSet("sse-check", flag.ContinueOnError) base := fs.String("base", "", "instance base URL") @@ -516,17 +566,41 @@ func cmdWrites(args []string) error { owner := fs.String("owner", "", "login owner") password := fs.String("password", "", "login password") conv := fs.String("conv", "", "conversation id") + project := fs.String("project", "", "project id used when -fresh creates a conversation per write") + fresh := fs.Bool("fresh", false, "one new conversation per write, so a live run cannot 409 the probe") n := fs.Int("n", 5, "consecutive successful writes required") if err := fs.Parse(args); err != nil { return err } + if *fresh && *project == "" { + return fmt.Errorf("writes -fresh requires -project") + } + if !*fresh && *conv == "" { + return fmt.Errorf("writes requires -conv") + } s, err := newSession(*base, *owner, *password) if err != nil { return err } for i := 0; i < *n; i++ { + target := *conv + if *fresh { + created, err := s.do("POST", "/api/projects/"+*project+"/conversations", map[string]any{ + "title": fmt.Sprintf("recover-%d", i), + }) + if err != nil { + return fmt.Errorf("write %d conversation: %w", i, err) + } + var body struct { + ID string `json:"id"` + } + if err := json.Unmarshal(created.Body, &body); err != nil || body.ID == "" { + return fmt.Errorf("write %d conversation id missing", i) + } + target = body.ID + } reqID := fmt.Sprintf("wr-%d-%d", time.Now().UnixNano(), i) - res, err := s.do("POST", "/api/conversations/"+*conv+"/messages", map[string]any{ + res, err := s.do("POST", "/api/conversations/"+target+"/messages", map[string]any{ "text": "recover " + reqID, "requestId": reqID, }) if err != nil || res.Status < 200 || res.Status >= 300 { diff --git a/tests/ops/drill_driver/window_test.go b/tests/ops/drill_driver/window_test.go new file mode 100644 index 00000000..553b6f8a --- /dev/null +++ b/tests/ops/drill_driver/window_test.go @@ -0,0 +1,21 @@ +package main + +import "testing" + +func TestAttemptInWindow(t *testing.T) { + cases := []struct { + started, since, until string + want bool + }{ + {"2026-10-12T13:37:55.091216802Z", "2026-10-12T13:37:57Z", "2026-10-12T13:38:57Z", false}, + {"2026-10-12T13:37:57Z", "2026-10-12T13:37:57Z", "2026-10-12T13:38:57Z", true}, + {"2026-10-12T13:38:57.000000001Z", "2026-10-12T13:37:57Z", "2026-10-12T13:38:57Z", false}, + {"2026-10-12T13:38:10.5Z", "", "", true}, + {"not-a-time", "2026-10-12T13:37:57Z", "2026-10-12T13:38:57Z", false}, + } + for _, tc := range cases { + if got := attemptInWindow(tc.started, tc.since, tc.until); got != tc.want { + t.Errorf("attemptInWindow(%s, %s, %s) = %v, want %v", tc.started, tc.since, tc.until, got, tc.want) + } + } +} diff --git a/tests/ops/run_fault_drill.sh b/tests/ops/run_fault_drill.sh index 0fbe74ee..f04e73c9 100755 --- a/tests/ops/run_fault_drill.sh +++ b/tests/ops/run_fault_drill.sh @@ -154,6 +154,8 @@ run_pg_phase() { prepare_conv "$phase/prep.json" local conv conv="$(python3 -c 'import json,sys; print(json.load(open(sys.argv[1]))["conversationId"])' "$phase/prep.json")" + local project + project="$(python3 -c 'import json,sys; print(json.load(open(sys.argv[1]))["projectId"])' "$phase/prep.json")" "$DRIVER" timeline -base "http://127.0.0.1:$PORT_A" -owner "$ADMIN_OWNER" -password "$ADMIN_PASS" \ -conv "$conv" -every 500ms -dur 100s -out "$phase/timeline.json" >"$phase/timeline.stdout" 2>&1 & local tl_pid=$! @@ -163,6 +165,8 @@ run_pg_phase() { else ops_pg_kill "$ROOT" "$NAME" fi + # 停机返回后留 2 秒，让已经在途的写入结束。断连窗口从这之后才开始计。 + sleep 2 local outage_start outage_end outage_start="$(date -u +%Y-%m-%dT%H:%M:%SZ)" # F2 与 60 秒断连并行：新实例必须拒绝启动，且不落 SQLite。 @@ -192,6 +196,9 @@ run_pg_phase() { wait "$(cat "$phase/probe.pid")" 2>/dev/null || true local sqlite_n sqlite_n="$(find "$probe_cwd" "$RESULTS/cwd-a" "$RESULTS/cwd-b" -name '*.db' 2>/dev/null | wc -l | tr -d ' ')" + local apps_alive=1 + kill -0 "$(head -1 "$APP_A_PIDFILE")" 2>/dev/null || apps_alive=0 + kill -0 "$(head -1 "$APP_B_PIDFILE")" 2>/dev/null || apps_alive=0 if [[ "$infra" -eq 1 ]]; then echo "INFRA_FAIL pg returned early" >"$phase/verdict.txt" note_fail "$mode infra" @@ -205,9 +212,10 @@ run_pg_phase() { local recovered=0 tries=0 while (( $(date +%s) - recover_start < 900 )); do tries=$((tries + 1)) - if "$DRIVER" writes -base "http://127.0.0.1:$PORT_A" -owner "$ADMIN_OWNER" -password "$ADMIN_PASS" -conv "$conv" -n 5 \ + # 时间线会话上可能还有 mock run。五次成功写到五个新会话，避免把 409 当成库没恢复。 + if "$DRIVER" writes -base "http://127.0.0.1:$PORT_A" -owner "$ADMIN_OWNER" -password "$ADMIN_PASS" -project "$project" -fresh -n 5 \ >"$phase/writes-a.txt" 2>&1 \ - && "$DRIVER" writes -base "http://127.0.0.1:$PORT_B" -owner "$ADMIN_OWNER" -password "$ADMIN_PASS" -conv "$conv" -n 5 \ + && "$DRIVER" writes -base "http://127.0.0.1:$PORT_B" -owner "$ADMIN_OWNER" -password "$ADMIN_PASS" -project "$project" -fresh -n 5 \ >"$phase/writes-b.txt" 2>&1; then recovered=1 break @@ -224,7 +232,8 @@ run_pg_phase() { local rto=$(( $(date +%s) - recover_start )) wait "$tl_pid" || true "$DRIVER" idem-verify -base "http://127.0.0.1:$PORT_A" -owner "$ADMIN_OWNER" -password "$ADMIN_PASS" \ - -conv "$conv" -in "$phase/timeline.json" -out "$phase/replays.json" || true + -conv "$conv" -in "$phase/timeline.json" -since "$outage_start" -until "$outage_end" \ + -out "$phase/replays.json" || true "$DRIVER" sse-check -base "http://127.0.0.1:$PORT_B" -owner "$ADMIN_OWNER" -password "$ADMIN_PASS" \ -conv "$conv" -out "$phase/sse-b.json" || true sleep 5 @@ -252,10 +261,10 @@ PY echo "orphan_runs=$(psql "$APP_DSN" -tAc "SELECT COUNT(*) FROM runs WHERE status IN ('pending','running','waiting_hitl')")" } >"$phase/invariants.txt" set +e - DRILL_DSN="$APP_DSN" python3 - "$phase" "$outage_start" "$outage_end" "$rto" "$recovered" "$recover_note" "$probe_alive" "$sqlite_n" <<'PY' + DRILL_DSN="$APP_DSN" python3 - "$phase" "$outage_start" "$outage_end" "$rto" "$recovered" "$recover_note" "$probe_alive" "$sqlite_n" "$apps_alive" <<'PY' import json, os, subprocess, sys from datetime import datetime -phase, outage_start, outage_end, rto, recovered, note, probe_alive, sqlite_n = sys.argv[1:] +phase, outage_start, outage_end, rto, recovered, note, probe_alive, sqlite_n, apps_alive = sys.argv[1:] dsn = os.environ["DRILL_DSN"] def parse(s): return datetime.fromisoformat(s.replace("Z", "+00:00")) @@ -266,8 +275,7 @@ slow = [a for a in window if a["elapsedMs"] > 10000] hangs = [a for a in window if a["status"] == 0 and a["elapsedMs"] >= 10000] ok_in_window = [a for a in window if 200 <= a["status"] < 300] late = ok_in_window -f1 = len(window) > 0 and not slow and not hangs and not late -apps_alive = True +f1 = len(window) > 0 and not slow and not hangs and not late and apps_alive == "1" f2 = probe_alive == "0" and sqlite_n == "0" and "拒绝就绪" in open(os.path.join(phase, "probe.log"), errors="replace").read() f3 = recovered == "1" and int(rto) <= 900 replays = json.load(open(os.path.join(phase, "replays.json"))) @@ -296,8 +304,9 @@ doc = { "F1": f1, "F2": f2, "F3": f3, "F4": f4, "F5": f5, "rto_s": int(rto), "recover_note": note, "probe_alive": probe_alive == "1", "sqlite_files": int(sqlite_n), "replay_bad": len(replay_bad), + "apps_alive": apps_alive == "1", "request_dups": dup.strip(), "missing_confirmed": missing, - "sse_holes": sse.get("holes"), "apps_alive": apps_alive, + "sse_holes": sse.get("holes"), } open(os.path.join(phase, "verdict.json"), "w").write(json.dumps(doc, indent=2)+"\n") print(json.dumps(doc, indent=2)) @@ -343,7 +352,7 @@ PY "$DRIVER" burst -bases "http://127.0.0.1:$PORT_A,http://127.0.0.1:$PORT_B" -owner "$USER_OWNER" -password "$USER_PASS" -n 40 -out "$phase/burst-shared.json" ops_redis_kill "$REDIS_DIR" "$REDIS_PORT" # G5：Redis 断连时新实例应拒绝启动。 - local probe_cwd="$phase/probe-cwd" + probe_cwd="$phase/probe-cwd" mkdir -p "$probe_cwd" set +e ( @@ -351,11 +360,11 @@ PY timeout 20 "$BIN" serve --dsn "$APP_DSN" --addr "127.0.0.1:$PROBE_PORT" --tier team --auth on --mock \ --instance u7drill-probe --artifacts "$ARTS" --disable-exec-tools >"$phase/probe.log" 2>&1 ) - local probe_ec=$? + probe_ec=$? set -e "$DRIVER" burst -bases "http://127.0.0.1:$PORT_A,http://127.0.0.1:$PORT_B" -owner "$USER_OWNER" -password "$USER_PASS" -n 40 -out "$phase/burst-down.json" prepare_conv "$phase/prep.json" || true - local conv="" + conv="" conv="$(python3 -c 'import json,sys; print(json.load(open(sys.argv[1])).get("conversationId",""))' "$phase/prep.json" 2>/dev/null || true)" if [[ -n "$conv" ]]; then "$DRIVER" writes -base "http://127.0.0.1:$PORT_A" -owner "$ADMIN_OWNER" -password "$ADMIN_PASS" -conv "$conv" -n 3 >"$phase/writes-down.txt" 2>&1 || true @@ -369,6 +378,7 @@ PY echo "$key $(redis-cli -p "$REDIS_PORT" pttl "$key")" >>"$phase/redis-pttl.txt" done <"$phase/redis-keys.txt" grep -n -i 'degraded\|redis' "$RESULTS/app-a.log" "$RESULTS/app-b.log" >"$phase/log-grep.txt" || true + set +e python3 - "$phase" "$probe_ec" <<'PY' import json, os, sys phase, probe_ec = sys.argv[1], int(sys.argv[2]) @@ -412,7 +422,9 @@ print(json.dumps(doc, indent=2)) if not (g1 and g2 and g3): sys.exit(1) PY - [[ $? -eq 0 ]] || note_fail "redis criteria" + judge=$? + set -e + [[ "$judge" -eq 0 ]] || note_fail "redis criteria" fi echo "[fault] done fail=$verdict_fail results=$RESULTS"