fix(library): 后台静默必须在关入口之后证明,并在停写者后复核
build / build (push) Successful in 1m19s

原顺序先证明后台空、再关入口排空:排空期间还在服务的在途请求会派生新任务,
那一刻的空到停写者时已经失效,而 app 收到 SIGTERM 会 exit(1) 把任务腰斩。

现在固定为:关入口并排空 → 证明后台静默(连续为空 + 稳定窗口 + 任务指纹)→
停写入者 → 停完再复核(队列非空/指纹变化即判 background_truncated)。
scheduled/retry 自行产生任务或任务卡住时 fail closed;旧 journal 的
background_idle 布尔不再当证明;入口关到一半失败时恢复入口并保持维护。
This commit is contained in:
姚章浩
2026-10-29 04:02:59 +08:00
parent c7dafe987e
commit 0ef79846e8
7 changed files with 792 additions and 66 deletions
+34
View File
@@ -1,5 +1,39 @@
# Changelog
## 开发快照(后台静默必须在关入口之后证明,且要能复核)
### Fixed
- **顺序缺口**:原顺序是「先证明后台静默 → 关入口排空 → 停写者」。排空期间还在服务的
在途请求随时会派生新的异步任务,那个"空"到停写者时早就失效;而 app 收到 SIGTERM 会
exit(1),正在跑的任务被腰斩。现在固定为「关入口并排空 → 证明后台静默 → 停写者 →
停完再复核」,并且在**入口关闭之前禁止取样**(`awaitBackgroundQuiet` 要求
`Gate.Closed && Gate.Drained`)。
- **单次瞬时 0 不算证明**:新证据 `QuiesceProof` 要求连续取样为空、跨过稳定窗口
(默认 1.5s)、任务集合指纹一致;只有 `Idle` 没有指纹的检查器在实例级直接拒绝。
- **停写者后复核**:`verifyBackgroundAfterStop` 再取一次样,队列非空或指纹变化即判
`background_truncated`,按"有任务被截断"失败,绝不继续写目标。
- **scheduled/retry 自行产生任务**:队列长时间没有进展(默认 30s)就直接失败
(`background_busy`),不拿等待时间冒充排空。
- **旧 journal 的 `background_idle` 布尔不再当证明**:只保留为历史字段,恢复流程一律
重新取样(`RecoverWindow` 不读它);只有布尔、没有证明的旧窗口按"未证明"处理。
- **恢复路径同样要求先停任务再停写者**:写入者已停但队列还有任务时立即失败并写清
外部处理方式(先起服务、入口保持关闭让任务跑完,再重试恢复),不再拿"服务没跑"
冒充"后台已静默"。
- **入口关到一半失败会把入口留在关闭状态**:`restore` 现在对实例级维护总是重新打开
入口(不再只看"是否成功关闭过"),放弃窗口时还要用 `ingressReady` 确认入口真的回来了,
否则保持维护不释放——避免"维护已释放、服务在跑、入口却是关的"。
- **zset 队列读取**:对 `scheduled`/`retry` 先 `TYPE` 再 `ZRANGE`,此前直接 `LRANGE`
会报 WRONGTYPE,任何自带重试/定时任务的实例都开不了窗口。
### Added
- `QuiesceSnapshot` / `QuiesceProof` 与 `StableQuiescer`:后台静默从"一次布尔"变成
可复核证据(样本数、稳定窗口、任务指纹、停写后复核时间)。
- `SetQuiesceStability(stable, stall)`:调整稳定窗口与"任务无进展"判定。
- 回归测试:取样必须晚于关入口、排空期间产生的任务要等它结束、队列一直有任务要
fail closed、停写后冒出新任务要判截断、旧布尔不算证明、只有 Idle 的检查器被拒。
## 开发快照(恢复校验按归档身份核对共享表)
### Fixed
+82 -3
View File
@@ -56,6 +56,44 @@ type Quiescer interface {
Idle(ctx context.Context) (bool, string, error)
}
// QuiesceSnapshot 是一次后台任务取样:队列深度 + 任务集合指纹。
// 指纹的作用是把"这一瞬间恰好是 0"和"一段时间内都没有新任务产生"区分开:
// scheduled/retry 会自己冒出新任务,只看瞬时 0 无法证明切点。
type QuiesceSnapshot struct {
Depth int64 `json:"depth"`
PendingOps int64 `json:"pending_ops"`
Unsettled int64 `json:"unsettled"`
Digest string `json:"digest"`
Detail string `json:"detail"`
}
// StableQuiescer 能给出任务集合指纹,才谈得上"证明切点"。
// 实例级维护必须用它;只会回答 Idle 的检查器只能证明某一瞬间是 0。
type StableQuiescer interface {
Quiescer
Snapshot(ctx context.Context) (QuiesceSnapshot, error)
}
// QuiesceProof 是"入口关掉之后后台真的一直是空的"的可复核证据:
//
// AfterIngress 之后才开始取样(排空期间在途请求还会派生新任务);
// 连续 Samples 次取样都为空,且跨过 StableMS 的时间窗;
// 停写入者之后再取一次样,Depth 仍为 0、指纹不变(否则就是切点不一致=有任务被截断)。
type QuiesceProof struct {
AfterIngress time.Time `json:"after_ingress"`
Samples int `json:"samples"`
StableMS int64 `json:"stable_ms"`
Depth int64 `json:"depth"`
PendingOps int64 `json:"pending_ops"`
Unsettled int64 `json:"unsettled"`
Digest string `json:"digest"`
Detail string `json:"detail"`
ProvenAt time.Time `json:"proven_at"`
PostStopDepth int64 `json:"post_stop_depth"`
PostStopDigest string `json:"post_stop_digest"`
PostStopAt time.Time `json:"post_stop_at"`
}
// Inventory must enumerate the complete, independently scoped test writer set,
// using immutable instance IDs. Never implement it as only the currently running
// subset. Start must wait for readiness; callers still read back the result.
@@ -97,8 +135,11 @@ type journal struct {
BackupSnapshotID string `json:"backup_snapshot_id,omitempty"`
// TargetFingerprint 是开窗口时的目标身份:恢复时必须还是同一个目标。
TargetFingerprint string `json:"target_fingerprint,omitempty"`
// BackgroundIdle 记录"停写入者之前后台任务队列确认为空"。
// BackgroundIdle 只记录"这个窗口曾经等到过空队列";它是历史事实,
// 不能当作"再次开入口/重启之后仍然静默"的证明——那种判断只认 Quiesce。
BackgroundIdle bool `json:"background_idle,omitempty"`
// Quiesce 是入口关闭之后的后台切点证据(稳定窗口 + 停写者后的复核)。
Quiesce *QuiesceProof `json:"quiesce,omitempty"`
// 恢复证据:只有真正跑完"恢复 + 校验"才会写这三项。
RecoveredBackup string `json:"recovered_backup,omitempty"`
RecoveryDigest string `json:"recovery_digest,omitempty"`
@@ -122,6 +163,10 @@ type Coordinator struct {
quiescer Quiescer
fingerprint string
quiesceWait time.Duration
// quiesceStable 是"连续为空必须跨过的时间窗":单次瞬时 0 不算证明。
quiesceStable time.Duration
// quiesceStall 是"同一批任务多久没有进展就判定卡住/被截断"。
quiesceStall time.Duration
}
func Open(root, source string, c Controller) (*Coordinator, error) {
@@ -194,6 +239,15 @@ func (c *Coordinator) SetQuiesceTimeout(d time.Duration) error {
return nil
}
// SetQuiesceStability 调整"稳定切点"的时间窗与任务停滞判定(默认 1.5s / 30s)。
func (c *Coordinator) SetQuiesceStability(stable, stall time.Duration) error {
if stable <= 0 || stall <= 0 {
return errors.New("invalid quiesce stability window")
}
c.quiesceStable, c.quiesceStall = stable, stall
return nil
}
// SetQuiescer 声明后台任务静默检查(HTTP 排空不等于后台任务排空)。
func (c *Coordinator) SetQuiescer(q Quiescer) error {
if q == nil {
@@ -408,6 +462,25 @@ func (c *Coordinator) load(job string) (journal, error) {
if j.BackupSnapshotID != "" && !validDigest(j.BackupSnapshotID) {
return journal{}, errors.New("invalid backup snapshot id")
}
// 后台静默只认带稳定窗口与停写后复核的证据。旧的 background_idle 布尔
// 只是历史记录:它不能证明"重启/再开入口之后仍然静默",也不能证明停写者
// 那一刻没有任务在跑。这里允许它存在(生存旧 journal),但绝不能当证明用:
// RecoverWindow 每次都会重新取样,不读这个布尔。
if j.BackgroundIdle && j.Quiesce == nil && !terminal(j.State) {
// 非终态窗口没有证明:状态上就是"还没证明静默",交给恢复流程重新取证。
j.BackgroundIdle = false
}
if q := j.Quiesce; q != nil {
if q.Samples < 2 || q.StableMS < 0 || q.Depth != 0 || q.PendingOps != 0 || q.Unsettled != 0 {
return journal{}, errors.New("invalid quiesce proof")
}
if !validDigest(q.Digest) || q.AfterIngress.IsZero() || q.ProvenAt.IsZero() || q.ProvenAt.Before(q.AfterIngress) {
return journal{}, errors.New("invalid quiesce proof")
}
if !q.PostStopAt.IsZero() && (q.PostStopDepth != 0 || q.PostStopDigest != q.Digest || q.PostStopAt.Before(q.ProvenAt)) {
return journal{}, errors.New("invalid post-stop quiesce proof")
}
}
// 恢复证据只允许指向"这次窗口自己那份预备份",且必须带校验摘要。
if j.RecoveredAt != nil {
if j.RecoveredBackup == "" || j.RecoveredBackup != j.BackupSnapshotID || j.RecoveryDigest == "" {
@@ -547,12 +620,18 @@ func (c *Coordinator) restore(ctx context.Context, j *journal) error {
}
}
// 服务都回来了才开放入口:先开门再启动服务,等于把流量打向没准备好的目标。
if j.Gate != nil {
//
// 注意:入口关到一半失败(闸门已被停、排空超时)时 j.Gate 可能还是 nil,
// 但入口确实已经不在了。所以实例级维护一律重新打开入口,而不是只看
// "有没有成功关闭的证据"——否则会出现"维护已释放、服务在跑、入口却是关的"。
if c.level == "instance" || j.Gate != nil {
if c.gate == nil {
failures = errors.Join(failures, errors.New("入口闸门缺失:不能确认目标已恢复"))
} else if err := c.gate.OpenIngress(ctx); err != nil {
failures = errors.Join(failures, fmt.Errorf("目标入口恢复失败:%w", err))
} else {
} else if j.Gate != nil {
// 只有"确实关过"的窗口才记"重新打开":没有关闭证据时开门是恢复原状,
// 不能写成"重开",否则 journal 校验会误判。
j.GateReopened = true
if err := c.save(*j); err != nil {
failures = errors.Join(failures, err)
@@ -0,0 +1,340 @@
package libraryfreeze
import (
"context"
"strings"
"testing"
"time"
)
// scriptedQuiescer 按脚本回答取样,并把每次取样记进共享调用序列:
// 顺序断言必须能看见"取样发生在关入口之后、停写者之前"。
type scriptedQuiescer struct {
calls *[]string
samples int
next func(n int) QuiesceSnapshot
}
func (q *scriptedQuiescer) Idle(ctx context.Context) (bool, string, error) {
snapshot, err := q.Snapshot(ctx)
if err != nil {
return false, "", err
}
idle := snapshot.Depth == 0 && snapshot.PendingOps == 0 && snapshot.Unsettled == 0
return idle, snapshot.Detail, nil
}
func (q *scriptedQuiescer) Snapshot(context.Context) (QuiesceSnapshot, error) {
q.samples++
*q.calls = append(*q.calls, "quiesce")
return q.next(q.samples), nil
}
func busySample(digest string) QuiesceSnapshot {
return QuiesceSnapshot{Depth: 1, Digest: strings.Repeat(digest, 64), Detail: "asynq pending=1"}
}
func quietSample() QuiesceSnapshot {
return QuiesceSnapshot{Depth: 0, Digest: strings.Repeat("e", 64), Detail: "asynq=0 pending_ops=0"}
}
func firstIndex(calls []string, want string) int {
for i, call := range calls {
if call == want {
return i
}
}
return -1
}
// 取样必须发生在入口确认关闭之后:关之前拿到的"空"会被排空期间的新任务推翻。
func TestQuiesceProofStartsOnlyAfterIngressClosed(t *testing.T) {
_, c, e := instanceFixture(t)
q := &scriptedQuiescer{calls: &c.calls, next: func(int) QuiesceSnapshot { return quietSample() }}
if err := e.SetQuiescer(q); err != nil {
t.Fatal(err)
}
if _, err := e.RunWindow(context.Background(), "job-order", func(context.Context) error { return nil },
func(context.Context, *Window) (string, error) { return "", nil }, func(error) bool { return true }); err != nil {
t.Fatal(err)
}
gateAt := firstIndex(c.calls, "gate:close")
stopAt := firstIndex(c.calls, "stop:app")
if gateAt < 0 || stopAt < 0 {
t.Fatalf("calls=%v", c.calls)
}
for i, call := range c.calls {
if call == "quiesce" && i < gateAt {
t.Fatalf("quiesce sampled before the ingress was closed: %v", c.calls)
}
}
proofProbes := 0
for i, call := range c.calls {
if call == "quiesce" && i < stopAt {
proofProbes++
}
}
if proofProbes < 2 {
t.Fatalf("only %d quiet samples before the writers were stopped: %v", proofProbes, c.calls)
}
afterStop := 0
for i, call := range c.calls {
if call == "quiesce" && i > stopAt {
afterStop++
}
}
if afterStop == 0 {
t.Fatalf("no post-stop verification sample: %v", c.calls)
}
journalAfter, err := e.load("job-order")
if err != nil || journalAfter.Quiesce == nil || journalAfter.Quiesce.Samples < 2 {
t.Fatalf("journal=%+v err=%v", journalAfter, err)
}
if journalAfter.Quiesce.PostStopAt.IsZero() {
t.Fatal("停写者之后没有复核后台任务集合")
}
}
// 排空期间通过真实入口派生出来的任务:窗口必须等它结束,不能直接停写者。
func TestWindowWaitsForTaskEnqueuedDuringDrain(t *testing.T) {
_, c, e := instanceFixture(t)
q := &scriptedQuiescer{calls: &c.calls, next: func(n int) QuiesceSnapshot {
if n <= 3 {
if firstIndex(c.calls, "stop:app") >= 0 {
t.Fatalf("writer stopped while a task was still running: %v", c.calls)
}
return busySample("b")
}
return quietSample()
}}
if err := e.SetQuiescer(q); err != nil {
t.Fatal(err)
}
if err := e.SetQuiesceStability(5*time.Millisecond, time.Minute); err != nil {
t.Fatal(err)
}
work := 0
if _, err := e.RunWindow(context.Background(), "job-drain-task", func(context.Context) error { return nil },
func(context.Context, *Window) (string, error) { work++; return "", nil }, func(error) bool { return true }); err != nil {
t.Fatal(err)
}
if work != 1 {
t.Fatalf("work=%d", work)
}
journalAfter, err := e.load("job-drain-task")
if err != nil {
t.Fatal(err)
}
proof := journalAfter.Quiesce
if proof == nil || proof.Depth != 0 || proof.PostStopDepth != 0 || proof.PostStopDigest != proof.Digest {
t.Fatalf("proof=%+v", proof)
}
if q.samples < 4 {
t.Fatalf("sampled %d times, expected to keep sampling while the task ran", q.samples)
}
}
// 队列一直不空(scheduled/retry 自己冒任务):宁可不进窗口,也不把任务留给 SIGTERM 腰斩。
func TestWindowFailsClosedWhenBackgroundKeepsProducing(t *testing.T) {
_, c, e := instanceFixture(t)
q := &scriptedQuiescer{calls: &c.calls, next: func(n int) QuiesceSnapshot {
return busySample([]string{"a", "b", "c"}[n%3])
}}
if err := e.SetQuiescer(q); err != nil {
t.Fatal(err)
}
if err := e.SetQuiesceTimeout(400 * time.Millisecond); err != nil {
t.Fatal(err)
}
if err := e.SetQuiesceStability(5*time.Millisecond, time.Minute); err != nil {
t.Fatal(err)
}
work := 0
_, err := e.RunWindow(context.Background(), "job-producer", func(context.Context) error { return nil },
func(context.Context, *Window) (string, error) { work++; return "", nil }, func(error) bool { return true })
if err == nil || !strings.Contains(err.Error(), "后台任务") {
t.Fatalf("err=%v", err)
}
if work != 0 {
t.Fatal("work ran although a producer kept the queue busy")
}
if firstIndex(c.calls, "stop:app") >= 0 {
t.Fatalf("writers were stopped while tasks kept arriving: %v", c.calls)
}
for _, s := range c.services {
if !s.Running {
t.Fatalf("service %s stayed stopped after the window was abandoned", s.ID)
}
}
journalAfter, loadErr := e.load("job-producer")
if loadErr != nil || journalAfter.State != "failed" || journalAfter.Code != "background_busy" {
t.Fatalf("journal=%+v err=%v", journalAfter, loadErr)
}
if journalAfter.Quiesce != nil {
t.Fatalf("failed window carries a fake quiet proof: %+v", journalAfter.Quiesce)
}
}
// 任务在"证明静默"和"停写者"之间冒出来:停完的复核必须发现它,按截断处理。
func TestWindowFailsClosedWhenTaskAppearsAfterWriterStopped(t *testing.T) {
_, c, e := instanceFixture(t)
q := &scriptedQuiescer{calls: &c.calls, next: func(int) QuiesceSnapshot {
if firstIndex(c.calls, "stop:app") >= 0 {
return busySample("c")
}
return quietSample()
}}
if err := e.SetQuiescer(q); err != nil {
t.Fatal(err)
}
work := 0
_, err := e.RunWindow(context.Background(), "job-late-task", func(context.Context) error { return nil },
func(context.Context, *Window) (string, error) { work++; return "", nil }, func(error) bool { return true })
if err == nil || !strings.Contains(err.Error(), "切点") {
t.Fatalf("err=%v", err)
}
if work != 0 {
t.Fatal("work ran with an inconsistent cut point")
}
journalAfter, loadErr := e.load("job-late-task")
if loadErr != nil || journalAfter.Code != "background_truncated" || journalAfter.State != "failed" {
t.Fatalf("journal=%+v err=%v", journalAfter, loadErr)
}
// 还没写目标:可以把服务恢复,但绝不能再往下写。
for _, s := range c.services {
if !s.Running {
t.Fatalf("service %s stayed stopped after an abandoned window", s.ID)
}
}
}
// 旧 journal 里的 background_idle 布尔仍然能读,但绝不当作"后台已静默"的证据。
func TestLegacyBackgroundIdleIsNotProof(t *testing.T) {
root, _, e := instanceFixture(t)
j := heldJournal("job-old-idle", strings.Repeat("a", 64), strings.Repeat("b", 64))
j.State = "stopping"
j.ContentRisk = false
j.RestoreBlocked = false
j.BackgroundIdle = true
if err := e.save(j); err != nil {
t.Fatal(err)
}
reopened, err := Open(root, "target-instance", e.control)
if err != nil {
t.Fatal(err)
}
defer reopened.Close()
loaded, err := reopened.load("job-old-idle")
if err != nil {
t.Fatalf("legacy journal became unreadable: %v", err)
}
if loaded.Quiesce != nil || loaded.BackgroundIdle {
t.Fatalf("legacy idle flag was promoted to a proof: %+v", loaded)
}
}
// 恢复要自己重新证明后台静默:journal 里的旧证明不能拿来复用。
func TestRecoverWindowReProvesBackgroundQuiesce(t *testing.T) {
_, c, e := instanceFixture(t)
stopAllServices(c)
if err := e.SetTargetFingerprint(strings.Repeat("a", 64)); err != nil {
t.Fatal(err)
}
j := heldJournal("job-stale-proof", strings.Repeat("a", 64), strings.Repeat("b", 64))
now := time.Now().UTC()
j.BackgroundIdle = true
j.Quiesce = &QuiesceProof{AfterIngress: now.Add(-time.Minute), Samples: 3, StableMS: 1500,
Digest: strings.Repeat("e", 64), Detail: "asynq=0", ProvenAt: now.Add(-30 * time.Second)}
if err := e.save(j); err != nil {
t.Fatal(err)
}
if _, err := e.load("job-stale-proof"); err != nil {
t.Fatalf("fixture journal is invalid: %v", err)
}
if err := e.SetQuiescer(&scriptedQuiescer{calls: &c.calls, next: func(int) QuiesceSnapshot {
return busySample("d")
}}); err != nil {
t.Fatal(err)
}
if err := e.SetQuiesceTimeout(300 * time.Millisecond); err != nil {
t.Fatal(err)
}
if err := e.SetQuiesceStability(5*time.Millisecond, time.Minute); err != nil {
t.Fatal(err)
}
called := false
if _, err := e.RecoverWindow(context.Background(), "job-stale-proof",
restorerFunc(func(context.Context, string) error { called = true; return nil }),
verifierFunc(func(context.Context, string) (string, error) { return "x", nil })); err == nil {
t.Fatal("recovery reused the journal's stale quiet proof")
}
if called {
t.Fatal("restore ran although the background was busy")
}
after, loadErr := e.load("job-stale-proof")
if loadErr != nil || terminal(after.State) {
t.Fatalf("journal=%+v err=%v", after, loadErr)
}
for _, s := range c.services {
if s.Running {
t.Fatal("service was started although recovery could not prove background quiet")
}
}
}
// 写入者已停但队列里还有任务:这些任务被截断过,服务不起来就永远排不掉——直接报出来。
func TestRecoverWindowRefusesWithTruncatedTasks(t *testing.T) {
_, c, e := instanceFixture(t)
stopAllServices(c)
if err := e.SetTargetFingerprint(strings.Repeat("a", 64)); err != nil {
t.Fatal(err)
}
if err := e.save(heldJournal("job-trunc", strings.Repeat("a", 64), strings.Repeat("b", 64))); err != nil {
t.Fatal(err)
}
if err := e.SetQuiescer(&scriptedQuiescer{calls: &c.calls, next: func(int) QuiesceSnapshot {
return busySample("f")
}}); err != nil {
t.Fatal(err)
}
called := false
_, err := e.RecoverWindow(context.Background(), "job-trunc",
restorerFunc(func(context.Context, string) error { called = true; return nil }),
verifierFunc(func(context.Context, string) (string, error) { return "x", nil }))
if err == nil || !strings.Contains(err.Error(), "截断") {
t.Fatalf("err=%v", err)
}
if called {
t.Fatal("restore ran with truncated tasks in the queue")
}
after, loadErr := e.load("job-trunc")
if loadErr != nil || after.State != "held" || after.Code != "background_truncated" || !after.RestoreBlocked {
t.Fatalf("journal=%+v err=%v", after, loadErr)
}
for _, s := range c.services {
if s.Running {
t.Fatal("service was started although truncated tasks remain")
}
}
}
// 恢复时拿不到任务指纹(只会回答 Idle 的检查器)就不许收尾:那是"瞬时 0",不是切点。
func TestInstanceWindowRejectsQuiescerWithoutSnapshots(t *testing.T) {
_, _, e := instanceFixture(t)
if err := e.SetQuiescer(idleOnlyQuiescer{}); err != nil {
t.Fatal(err)
}
work := 0
_, err := e.RunWindow(context.Background(), "job-idle-only", func(context.Context) error { return nil },
func(context.Context, *Window) (string, error) { work++; return "", nil }, func(error) bool { return true })
if err == nil || !strings.Contains(err.Error(), "指纹") {
t.Fatalf("err=%v", err)
}
if work != 0 {
t.Fatal("window opened without a provable cut point")
}
}
type idleOnlyQuiescer struct{}
func (idleOnlyQuiescer) Idle(context.Context) (bool, string, error) { return true, "asynq=0", nil }
+22 -2
View File
@@ -37,6 +37,9 @@ func (c *Coordinator) RecoverWindow(ctx context.Context, job string, restorer Re
}
defer unlock(lock)
if c.level == "instance" && c.gate == nil {
return Result{}, errors.New("实例级恢复缺少入口闸门:不能证明入口已关闭")
}
j, err := c.load(job)
if err != nil {
return Result{}, err
@@ -71,6 +74,23 @@ func (c *Coordinator) RecoverWindow(ctx context.Context, job string, restorer Re
return j.Result, err
}
}
// 写入者已经停住但队列还有任务:这些任务在上一次中断里被截断,服务不起来就永远排不掉。
// 直接失败并写清外部处理方式,不拿"服务没跑"冒充"后台已静默"。
running, err := c.anyServiceRunning(ctx, j)
if err != nil {
return j.Result, err
}
if !running {
if stuck := c.checkBackgroundStuck(ctx); stuck != nil {
return c.failRecovery(&j, "background_truncated", stuck)
}
}
// 入口已经关了才证明后台静默:恢复期间关闭的入口不会派生新任务,
// 但上次中断留下的在跑任务必须等它结束,否则恢复后的服务一起来它就会继续写。
proof, quietErr := c.awaitBackgroundQuiet(ctx, &j)
if quietErr != nil {
return c.failRecovery(&j, "background_busy", fmt.Errorf("恢复前后台任务未静默:%w", quietErr))
}
if err := c.stopAll(ctx, &j); err != nil {
return j.Result, fmt.Errorf("恢复前无法停下写入者:%w", err)
}
@@ -80,8 +100,8 @@ func (c *Coordinator) RecoverWindow(ctx context.Context, job string, restorer Re
if err := c.verifyDrain(ctx, &j); err != nil {
return j.Result, fmt.Errorf("恢复前排空未确认:%w", err)
}
if err := c.awaitBackgroundIdle(ctx, &j); err != nil {
return j.Result, fmt.Errorf("恢复前后台任务未静默:%w", err)
if err := c.verifyBackgroundAfterStop(ctx, &j, proof); err != nil {
return c.failRecovery(&j, "background_truncated", fmt.Errorf("恢复前切点不一致:%w", err))
}
j.State = "recovering"
if err := c.save(j); err != nil {
+172 -25
View File
@@ -100,6 +100,11 @@ func (c *Coordinator) RunWindow(ctx context.Context, job string, prepare func(co
return Result{}, err
}
defer unlock(lock)
// 实例级维护的入口必须由我们关:没有闸门就没有"入口已关"的证据,
// 后面拿什么证明后台静默都是空的。
if c.level == "instance" && c.gate == nil {
return Result{}, errors.New("实例级维护缺少入口闸门:不能证明入口已关闭")
}
previous, loadErr := c.load(job)
j := journal{}
@@ -144,14 +149,9 @@ func (c *Coordinator) RunWindow(ctx context.Context, job string, prepare func(co
}
}
// 后台任务必须先在"还能跑"的时候跑完:入口排空只覆盖 HTTP 连接,
// app 内的 asynq worker 会被上游的 exit(1) 直接切断。
if idleErr := c.awaitBackgroundIdle(ctx, &j); idleErr != nil {
return c.abandonWindow(ctx, &j, idleErr, "background_busy")
}
// 顺序是有原因的:入口先停(新请求被拒、在途请求跑完),写者后停。
// 反过来做,正在处理的请求就会被 SIGTERM 切断——那不算排空。
// 顺序不能换:先把外部任务的生产入口关掉并排空,再证明后台静默,最后才停写入者。
// 先证明静默再关入口是错的——排空期间还在跑的在途请求随时会派生新的异步任务,
// 那一刻的"空"到停写者时早就失效了;而 app 一旦 SIGTERM 就 exit(1),会把任务腰斩。
if c.gate != nil && (j.Gate == nil || !j.Gate.Drained) {
evidence, gateErr := c.gate.CloseIngress(ctx)
if gateErr != nil {
@@ -166,6 +166,12 @@ func (c *Coordinator) RunWindow(ctx context.Context, job string, prepare func(co
}
}
// 入口关了才谈得上"稳定切点":连续为空 + 跨过稳定窗口 + 拿得到任务指纹。
proof, quietErr := c.awaitBackgroundQuiet(ctx, &j)
if quietErr != nil {
return c.abandonWindow(ctx, &j, quietErr, "background_busy")
}
// 停写入者:任何一步不成立都不许往下写。
if stopErr := c.stopAll(ctx, &j); stopErr != nil {
return c.abandonWindow(ctx, &j, stopErr, "stop_failed")
@@ -176,6 +182,11 @@ func (c *Coordinator) RunWindow(ctx context.Context, job string, prepare func(co
if stopErr := c.verifyDrain(ctx, &j); stopErr != nil {
return c.abandonWindow(ctx, &j, stopErr, "drain_unverified")
}
// 停完再复核一次:入队/派发可能在"证明静默"和"停写者"之间发生。
// 队列仍有任务说明有任务被截断——这比同步失败严重,按切点不一致处理,绝不往下写。
if stopErr := c.verifyBackgroundAfterStop(ctx, &j, proof); stopErr != nil {
return c.abandonWindow(ctx, &j, stopErr, "background_truncated")
}
if c.ingressClosed != nil {
if stopErr := c.ingressClosed(ctx); stopErr != nil {
return c.abandonWindow(ctx, &j, fmt.Errorf("目标入口仍然可用:%w", stopErr), "ingress_open")
@@ -228,6 +239,15 @@ func (c *Coordinator) abandonWindow(ctx context.Context, j *journal, cause error
_ = c.save(*j)
return j.Result, errors.Join(cause, restoreErr)
}
// 入口必须真的回来了才算"安全放弃":服务在跑但入口关着,等于把目标静默下线。
if c.ingressReady != nil {
if readyErr := c.ingressReady(recoveryCtx); readyErr != nil {
j.Code = code
j.State = "held"
_ = c.save(*j)
return j.Result, errors.Join(cause, fmt.Errorf("目标入口没有恢复:%w", readyErr))
}
}
ended := time.Now().UTC()
j.State = "failed"
j.Code = code
@@ -292,42 +312,128 @@ func (c *Coordinator) closeWindow(ctx context.Context, j *journal, cause error,
return j.Result, cause
}
// awaitBackgroundIdle 等到后台任务队列为空再往下走。拿不到结论(查询失败)
// 也按"没空"处理:不能把"问不到"当成"没任务"。
func (c *Coordinator) awaitBackgroundIdle(ctx context.Context, j *journal) error {
// awaitBackgroundQuiet 证明"入口关掉之后后台一直没任务",返回可复核证据。
//
// 关键点:
// - 必须在入口关闭且排空之后调用。排空期间在途请求还会派生任务,
// 在关入口之前取样得到的 0 到停写者时可能已失效。
// - 只认"连续为空 + 跨过稳定窗口 + 两次指纹一致",单次瞬时 0 不算证明。
// - 队列一直不空(scheduled/retry 自己冒任务、任务卡住)就失败:宁可不开窗口,
// 也不能把正在跑的任务留给 SIGTERM 腰斩。
func (c *Coordinator) awaitBackgroundQuiet(ctx context.Context, j *journal) (*QuiesceProof, error) {
if c.quiescer == nil {
if c.level == "instance" {
return errors.New("没有后台任务静默检查:不能证明停写者时后台任务已结束")
return nil, errors.New("没有后台任务静默检查:不能证明停写者时后台任务已结束")
}
return nil
return nil, nil
}
stable := c.quiesceStable
if stable <= 0 {
stable = 1500 * time.Millisecond
}
stall := c.quiesceStall
if stall <= 0 {
stall = 30 * time.Second
}
wait := c.quiesceWait
if wait <= 0 {
wait = 2 * time.Minute
}
deadline := time.Now().Add(wait)
var last string
st, stableOK := c.quiescer.(StableQuiescer)
if !stableOK {
if c.level == "instance" {
return nil, errors.New("后台任务检查不给任务集合指纹:无法证明稳定切点(scheduled/retry 可能自行产生)")
}
return nil, nil
}
if j.Gate == nil || !j.Gate.Drained {
return nil, errors.New("必须先关闭并排空入口,再证明后台静默")
}
started := time.Now().UTC()
deadline := started.Add(wait)
var anchor QuiesceSnapshot
var anchorAt time.Time
var lastDetail string
samples := 0
var busyDigest string
var busySince time.Time
for {
idle, detail, err := c.quiescer.Idle(ctx)
snapshot, err := st.Snapshot(ctx)
if err != nil {
return fmt.Errorf("读不到后台任务状态:%w", err)
return nil, fmt.Errorf("读不到后台任务状态:%w", err)
}
if idle {
j.BackgroundIdle = true
return nil
samples++
lastDetail = snapshot.Detail
now := time.Now().UTC()
if snapshot.Depth == 0 && snapshot.PendingOps == 0 && snapshot.Unsettled == 0 {
if anchorAt.IsZero() {
anchor, anchorAt = snapshot, now
} else if snapshot.Digest == anchor.Digest && now.Sub(anchorAt) >= stable {
proof := &QuiesceProof{
AfterIngress: started, Samples: samples, StableMS: now.Sub(anchorAt).Milliseconds(),
Depth: 0, PendingOps: 0, Unsettled: 0,
Digest: snapshot.Digest, Detail: snapshot.Detail, ProvenAt: now,
}
j.Quiesce = proof
j.BackgroundIdle = true
if err := c.save(*j); err != nil {
return nil, err
}
return proof, nil
}
} else {
anchor, anchorAt = QuiesceSnapshot{}, time.Time{}
if busyDigest != snapshot.Digest {
busyDigest, busySince = snapshot.Digest, now
} else if now.Sub(busySince) >= stall {
// 同一批任务长时间没有进展:被截断或需要外部处理,直接失败,不猜。
return nil, fmt.Errorf("后台任务没有进展(%s):先让它们结束再开窗口,否则会被 SIGTERM 切断", lastDetail)
}
}
last = detail
if time.Now().After(deadline) {
return fmt.Errorf("后台任务仍在运行(%s):等它结束再开窗口,否则会被 SIGTERM 切断", last)
if now.After(deadline) {
return nil, fmt.Errorf("后台任务仍在运行(%s):等它结束再开窗口,否则会被 SIGTERM 切断", lastDetail)
}
select {
case <-ctx.Done():
return ctx.Err()
case <-time.After(500 * time.Millisecond):
return nil, ctx.Err()
case <-time.After(200 * time.Millisecond):
}
}
}
// verifyBackgroundAfterStop 复核"停写者那一刻没有任务在跑":
// 队列在停完之后仍为空、指纹不变,才算切点一致;否则说明有任务被截断。
func (c *Coordinator) verifyBackgroundAfterStop(ctx context.Context, j *journal, proof *QuiesceProof) error {
if c.quiescer == nil {
return nil
}
if proof == nil {
if c.level == "instance" {
return errors.New("没有后台静默证据:不能声称停写者时后台已空")
}
return nil
}
st, ok := c.quiescer.(StableQuiescer)
if !ok {
return nil
}
snapshot, err := st.Snapshot(ctx)
if err != nil {
return fmt.Errorf("停写者后读不到后台任务状态:%w", err)
}
if snapshot.Depth != 0 || snapshot.PendingOps != 0 || snapshot.Unsettled != 0 || snapshot.Digest != proof.Digest {
return fmt.Errorf("停写入者后后台任务集合变了(%s):切点不一致,按任务被截断处理", snapshot.Detail)
}
proof.PostStopDepth = snapshot.Depth
proof.PostStopDigest = snapshot.Digest
proof.PostStopAt = time.Now().UTC()
if err := c.save(*j); err != nil {
return err
}
return nil
}
// stopAll 只对"原本在跑"的服务发停止指令,每发一个都先落盘意图:
// 进程在停止过程中死掉时,磁盘上仍能看出"哪些服务被要求停过"。
func (c *Coordinator) stopAll(ctx context.Context, j *journal) error {
@@ -415,3 +521,44 @@ func validDigest(id string) bool {
decoded, err := hex.DecodeString(id)
return err == nil && len(decoded) == 32 && strings.ToLower(id) == id
}
// anyServiceRunning 判断这个窗口记录的写入者里是否还有在跑的。
func (c *Coordinator) anyServiceRunning(ctx context.Context, j journal) (bool, error) {
items, err := inventory(ctx, c.control)
if err != nil {
return false, err
}
running := map[string]bool{}
for _, s := range items {
running[s.ID] = s.Running
}
for _, s := range j.Services {
if running[s.ID] {
return true, nil
}
}
return false, nil
}
// checkBackgroundStuck 在"写入者已经停住"的前提下判断后台是否还有没跑完的任务:
// 这时候没人能消费队列,等下去没有意义,只能报出来交给外部处理。
func (c *Coordinator) checkBackgroundStuck(ctx context.Context) error {
if c.quiescer == nil {
return nil
}
st, ok := c.quiescer.(StableQuiescer)
if !ok {
if c.level == "instance" {
return errors.New("后台任务检查不给任务集合指纹:无法判断是否有被截断的任务")
}
return nil
}
snapshot, err := st.Snapshot(ctx)
if err != nil {
return fmt.Errorf("读不到后台任务状态:%w", err)
}
if snapshot.Depth == 0 && snapshot.PendingOps == 0 && snapshot.Unsettled == 0 {
return nil
}
return fmt.Errorf("写入者已停但后台仍有任务(%s):这些任务在上次中断时被截断,需要先把服务起来(入口保持关闭)让它们跑完,再重试恢复", snapshot.Detail)
}
+37 -11
View File
@@ -30,6 +30,7 @@ func (c *exitControl) Inventory(ctx context.Context) ([]Service, error) {
}
// idleQuiescer:实例级窗口必须能证明后台任务已空,测试里默认给一个空的。
// 它同时实现 StableQuiescer:只有能给出任务集合指纹的检查器才谈得上证明切点。
type idleQuiescer struct{ busy bool }
func (q *idleQuiescer) Idle(context.Context) (bool, string, error) {
@@ -39,6 +40,13 @@ func (q *idleQuiescer) Idle(context.Context) (bool, string, error) {
return true, "asynq=0 pending_ops=0", nil
}
func (q *idleQuiescer) Snapshot(context.Context) (QuiesceSnapshot, error) {
if q.busy {
return QuiesceSnapshot{Depth: 1, Digest: strings.Repeat("b", 64), Detail: "asynq pending=1"}, nil
}
return QuiesceSnapshot{Depth: 0, Digest: strings.Repeat("e", 64), Detail: "asynq=0 pending_ops=0"}, nil
}
func instanceFixture(t *testing.T) (string, *exitControl, *Coordinator) {
t.Helper()
root := t.TempDir()
@@ -58,6 +66,13 @@ func instanceFixture(t *testing.T) (string, *exitControl, *Coordinator) {
if err := engine.SetQuiescer(&idleQuiescer{}); err != nil {
t.Fatal(err)
}
// 实例级窗口必须有入口闸门:默认装一个会正常排空的假闸门。
if err := engine.SetGate(&fakeGate{calls: &c.calls, drain: true, active: 1}); err != nil {
t.Fatal(err)
}
if err := engine.SetQuiesceStability(5*time.Millisecond, 200*time.Millisecond); err != nil {
t.Fatal(err)
}
return root, c, engine
}
@@ -95,27 +110,31 @@ func TestWindowStopsWritersBeforeWorkAndRestoresAfter(t *testing.T) {
if !j.DrainVerified || !j.IngressClosed || !j.ContentRisk || j.Level != "instance" {
t.Fatalf("journal evidence missing: %+v", j)
}
if strings.Join(c.calls, ",") != "stop:app,stop:store,start:app,start:store" {
if strings.Join(c.calls, ",") != "gate:close,stop:app,stop:store,start:app,start:store,gate:open" {
t.Fatalf("calls=%v", c.calls)
}
}
// 没有优雅退出证据(退出码非 0)时不许开始写目标:在途操作可能被切断。
func TestWindowRefusesWorkWhenDrainUnverified(t *testing.T) {
// 实例级窗口必须有入口闸门:没有闸门就无法证明"入口已关",后台静默也就没有前提。
// 拒绝发生在动任何容器之前。
func TestInstanceWindowRequiresGate(t *testing.T) {
_, c, e := instanceFixture(t)
c.exit["app"] = 143
e.gate = nil
work := 0
_, err := e.RunWindow(context.Background(), "job-drain", func(context.Context) error { return nil },
_, err := e.RunWindow(context.Background(), "job-nogate", func(context.Context) error { return nil },
func(context.Context, *Window) (string, error) { work++; return "", nil }, func(error) bool { return true })
if err == nil || !strings.Contains(err.Error(), "排空") {
if err == nil || !strings.Contains(err.Error(), "入口闸门") {
t.Fatalf("err=%v", err)
}
if work != 0 {
t.Fatal("work ran without drain evidence")
t.Fatal("work ran without an ingress gate")
}
if len(c.calls) != 0 {
t.Fatalf("containers touched without a gate: %v", c.calls)
}
for _, s := range c.services {
if !s.Running {
t.Fatal("writers were not restored after drain failure")
t.Fatal("writers were stopped without a gate")
}
}
}
@@ -212,7 +231,7 @@ func TestWindowCrashDuringWriteKeepsTargetHeld(t *testing.T) {
if err = reopened.Recover(context.Background()); err == nil {
t.Fatal("recovery released a target with unverified content")
}
if len(c.calls) != 2 {
if len(c.calls) != 3 || c.calls[0] != "gate:close" {
t.Fatalf("calls=%v", c.calls)
}
}
@@ -237,7 +256,7 @@ func TestWindowAdoptsStoppingJournal(t *testing.T) {
if err != nil {
t.Fatal(err)
}
if strings.Join(c.calls, ",") != "stop:store,start:app,start:store" {
if strings.Join(c.calls, ",") != "gate:close,stop:store,start:app,start:store,gate:open" {
t.Fatalf("calls=%v", c.calls)
}
}
@@ -321,9 +340,16 @@ func TestWindowRefusesWhenGateNotDrained(t *testing.T) {
if err == nil || work != 0 {
t.Fatalf("err=%v work=%d", err, work)
}
if strings.Join(c.calls, ",") != "gate:close" {
// 写入者一个都不许停;但入口已经被关过一次,放弃窗口时必须把它重新打开,
// 否则会出现"维护已释放、服务在跑、入口却是关的"。
if strings.Join(c.calls, ",") != "gate:close,gate:open" {
t.Fatalf("writers were stopped before the ingress was drained: %v", c.calls)
}
for _, s := range c.services {
if !s.Running {
t.Fatalf("service %s was stopped although the ingress never drained", s.ID)
}
}
}
// 窗口开过之后被人工拉起来的写入者,恢复时必须再停一次:
+105 -25
View File
@@ -3,14 +3,20 @@ package librarymaintenance
import (
"bufio"
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"net"
"sort"
"strconv"
"strings"
"time"
"github.com/jackc/pgx/v5/pgxpool"
"knowledge-governance/internal/libraryfreeze"
)
// targetQuiescer 回答"后台写入者是不是已经空了":
@@ -45,55 +51,96 @@ func newTargetQuiescer(ctx context.Context, databaseURL, redisAddr, redisPass st
func (q *targetQuiescer) Close() { q.pool.Close() }
func (q *targetQuiescer) Idle(ctx context.Context) (bool, string, error) {
// Snapshot 是一次"后台任务集合"取样:既给出深度,也给出任务指纹。
//
// 指纹是把"这一瞬间恰好空"和"一段时间内没有新任务产生"分开的唯一手段:
// scheduled/retry 会自己冒出新任务,只看瞬时 0 无法证明停写者时的切点。
func (q *targetQuiescer) Snapshot(ctx context.Context) (libraryfreeze.QuiesceSnapshot, error) {
var pendingOps, unsettled int64
// task_pending_ops / knowledges 是 app 自己的后台账本:不看它们只看 HTTP 会漏。
// task_pending_ops / pending_subtasks_count 是 app 自己的"正在做"账本:
// 不看它们只看 HTTP 会漏。注意不能拿 parse_status<>'completed' 当"在跑"——
// 解析失败/卡住的文档会永远停在那里,那是数据状态,不是会跑动的任务;
// 用它当判据会让维护窗口永久开不了(真跑踩过)。
if err := q.pool.QueryRow(ctx, `SELECT
(SELECT count(*) FROM task_pending_ops),
(SELECT count(*) FROM knowledges WHERE deleted_at IS NULL
AND (parse_status IS DISTINCT FROM 'completed' OR pending_subtasks_count IS NULL OR pending_subtasks_count<>0))`).
AND coalesce(pending_subtasks_count,0) <> 0)`).
Scan(&pendingOps, &unsettled); err != nil {
return false, "", fmt.Errorf("读不到后台账本:%w", err)
return libraryfreeze.QuiesceSnapshot{}, fmt.Errorf("读不到后台账本:%w", err)
}
depth, err := q.queueDepth(ctx)
tasks, err := q.queueTasks(ctx)
if err != nil {
return libraryfreeze.QuiesceSnapshot{}, err
}
sort.Strings(tasks)
sum := sha256.Sum256([]byte(strings.Join(tasks, "\n")))
return libraryfreeze.QuiesceSnapshot{
Depth: int64(len(tasks)),
PendingOps: pendingOps,
Unsettled: unsettled,
Digest: hex.EncodeToString(sum[:]),
Detail: fmt.Sprintf("pending_ops=%d unsettled_knowledges=%d asynq_tasks=%d",
pendingOps, unsettled, len(tasks)),
}, nil
}
func (q *targetQuiescer) Idle(ctx context.Context) (bool, string, error) {
snapshot, err := q.Snapshot(ctx)
if err != nil {
return false, "", err
}
detail := fmt.Sprintf("pending_ops=%d unsettled_knowledges=%d asynq_active=%d", pendingOps, unsettled, depth)
return pendingOps == 0 && unsettled == 0 && depth == 0, detail, nil
idle := snapshot.Depth == 0 && snapshot.PendingOps == 0 && snapshot.Unsettled == 0
return idle, snapshot.Detail, nil
}
// queueDepth 统计 asynq 里"还没结束"的任务数(排队/在跑/待重试)。
// queueTasks 列出"还没结束"的 asynq 任务(排队/在跑/待重试),每项是 键|任务ID。
// archived/completed 是历史,不算忙。
func (q *targetQuiescer) queueDepth(ctx context.Context) (int64, error) {
func (q *targetQuiescer) queueTasks(ctx context.Context) ([]string, error) {
conn, err := dialRedis(q.redisAddr, q.redisPass, q.redisDB, q.timeout)
if err != nil {
return 0, fmt.Errorf("连不上目标队列:%w", err)
return nil, fmt.Errorf("连不上目标队列:%w", err)
}
defer conn.Close()
keys, err := conn.scan(ctx, q.queueMatch)
if err != nil {
return 0, fmt.Errorf("扫描目标队列失败:%w", err)
return nil, fmt.Errorf("扫描目标队列失败:%w", err)
}
var total int64
sort.Strings(keys)
tasks := make([]string, 0, 32)
for _, key := range keys {
suffix := key[strings.LastIndex(key, ":"):]
switch suffix {
case ":pending", ":active", ":inprogress":
n, err := conn.intCmd(ctx, "LLEN", key)
if err != nil {
return 0, err
}
total += n
case ":scheduled", ":retry":
n, err := conn.intCmd(ctx, "ZCARD", key)
if err != nil {
return 0, err
}
total += n
case ":pending", ":inprogress", ":active", ":scheduled", ":retry":
default:
continue
}
items, err := conn.rangeAll(ctx, key)
if err != nil {
return nil, fmt.Errorf("读取目标队列 %s 失败:%w", key, err)
}
for _, item := range items {
tasks = append(tasks, key+"|"+taskIdentity(item))
}
}
return total, nil
return tasks, nil
}
// taskIdentity 从 asynq 的任务 JSON 里取稳定 ID;取不到就用整条内容摘要,
// 目的都是"同一批任务 -> 同一个指纹"。
func taskIdentity(raw string) string {
var payload map[string]json.RawMessage
if err := json.Unmarshal([]byte(raw), &payload); err == nil {
for _, key := range []string{"ID", "id"} {
if value, ok := payload[key]; ok {
var id string
if err := json.Unmarshal(value, &id); err == nil && id != "" {
return id
}
}
}
}
sum := sha256.Sum256([]byte(raw))
return hex.EncodeToString(sum[:])
}
// ---- 最小 RESP 客户端:只用到 AUTH/SELECT/SCAN/LLEN/ZCARD,不引第三方库 ----
@@ -129,6 +176,39 @@ func dialRedis(addr, password string, db int, timeout time.Duration) (*redisConn
func (c *redisConn) Close() error { return c.conn.Close() }
// rangeAll 读整个 list/zset。必须先问 TYPE 再选命令:
// 对 zset 发 LRANGE 会返回错误,而不是"不是 list"的普通回复。
func (c *redisConn) rangeAll(ctx context.Context, key string) ([]string, error) {
kind, err := c.do("TYPE", key)
if err != nil {
return nil, err
}
var command string
switch kind {
case "list":
command = "LRANGE"
case "zset":
command = "ZRANGE"
default:
return nil, nil
}
reply, err := c.do(command, key, "0", "-1")
if err != nil {
return nil, err
}
items, ok := reply.([]any)
if !ok {
return nil, errors.New("unexpected queue reply")
}
out := make([]string, 0, len(items))
for _, item := range items {
if text, ok := item.(string); ok {
out = append(out, text)
}
}
return out, nil
}
func (c *redisConn) do(args ...string) (any, error) {
_ = c.conn.SetDeadline(time.Now().Add(c.timeout))
var builder strings.Builder