diff --git a/CHANGELOG.md b/CHANGELOG.md index 1a327ad..33c5d00 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/internal/libraryfreeze/freeze.go b/internal/libraryfreeze/freeze.go index cc5643c..7077123 100644 --- a/internal/libraryfreeze/freeze.go +++ b/internal/libraryfreeze/freeze.go @@ -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) diff --git a/internal/libraryfreeze/quiesce_order_test.go b/internal/libraryfreeze/quiesce_order_test.go new file mode 100644 index 0000000..d917de6 --- /dev/null +++ b/internal/libraryfreeze/quiesce_order_test.go @@ -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 } diff --git a/internal/libraryfreeze/recovery.go b/internal/libraryfreeze/recovery.go index 360a91d..3f38f4a 100644 --- a/internal/libraryfreeze/recovery.go +++ b/internal/libraryfreeze/recovery.go @@ -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 { diff --git a/internal/libraryfreeze/window.go b/internal/libraryfreeze/window.go index 17773c9..f8dac27 100644 --- a/internal/libraryfreeze/window.go +++ b/internal/libraryfreeze/window.go @@ -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) +} diff --git a/internal/libraryfreeze/window_test.go b/internal/libraryfreeze/window_test.go index f352fe2..f6e1ba8 100644 --- a/internal/libraryfreeze/window_test.go +++ b/internal/libraryfreeze/window_test.go @@ -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) + } + } } // 窗口开过之后被人工拉起来的写入者，恢复时必须再停一次： diff --git a/internal/librarymaintenance/quiesce.go b/internal/librarymaintenance/quiesce.go index c78e010..9a4cfdf 100644 --- a/internal/librarymaintenance/quiesce.go +++ b/internal/librarymaintenance/quiesce.go @@ -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