weichen/Agent-System-Dev
1
0
Fork 0
Code Issues 1 Pull Requests 5 Actions 2 Packages Projects Releases Wiki Activity

Measure C4 recovery on fresh conversations

Browse Source 
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 <cursoragent@cursor.com>
This commit is contained in:
weichen
2026-10-12 14:20:13 +00:00
co-authored by Cursor
parent 4c2e326c03
commit de6eb084e5
3 changed files with 128 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/ops/drill_driver/main.go
+83 -9
View File
@@ -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 {
tests/ops/drill_driver/window_test.go
+21
View File
@@ -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)
}
}
}
tests/ops/run_fault_drill.sh
+24 -12
View File
@@ -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"