Blizzard
0378a770ca
live NATS 联调发现：resume 记录键用 "pending:"+taskID，冒号是 NATS JetStream KV 非法字符（仅允许 [-/_=.a-zA-Z0-9]）→ 中断时 persistResume 的 Put 静默失败、决定 到达时 loadResume 报 "nats: invalid key"，任务永卡 waiting、无法恢复。内存桩接受 任意键，故单测漏过——正是只有 live NATS 才暴露的那类。 - pendingKey: "pending:"+id → "pending_"+id（合法键）。 - persistResume: Put 失败改 log.Printf 大声告警（不止 exec 轨迹），关键失败可见。 - 回归测试 TestPendingKeyIsNATSValid：直接钉键形匹配 NATS KV 字符集，绕开内存桩盲区。 live 验证（devnats + 真链路）：提交 HITL 任务→waiting→杀 dispatcher→离线批准→ 重启→1s waiting→2s running→3s done，deepseek 真实出稿。证明 checkpoint 抗重启 + 决定经 JetStream 抗离线 + 断点恢复。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
81 lines
2.3 KiB
Go
81 lines
2.3 KiB
Go
package eino
|
||
|
||
import (
|
||
"context"
|
||
"regexp"
|
||
"sync"
|
||
"testing"
|
||
|
||
"github.com/sundynix/sundynix-shared/bus"
|
||
)
|
||
|
||
// natsKVKey 是 NATS JetStream KV 允许的键字符集（首字符不可为 . 或 _）。
|
||
// resume 记录键曾用冒号分隔 → 真 NATS 报 "invalid key"，内存桩测不到，故在此直接钉键形。
|
||
var natsKVKey = regexp.MustCompile(`^[a-zA-Z0-9][-/_=.a-zA-Z0-9]*$`)
|
||
|
||
func TestPendingKeyIsNATSValid(t *testing.T) {
|
||
for _, id := range []string{"task_4c47de5ce1a29e16", "abc123", "t_0"} {
|
||
if k := pendingKey(id); !natsKVKey.MatchString(k) {
|
||
t.Fatalf("pendingKey(%q)=%q 不是合法 NATS KV 键（冒号等字符会被拒）", id, k)
|
||
}
|
||
}
|
||
}
|
||
|
||
// memKV 是 CheckpointKV 的内存桩（并发安全），用于不依赖 NATS 的单测。
|
||
type memKV struct {
|
||
mu sync.Mutex
|
||
m map[string][]byte
|
||
}
|
||
|
||
func newMemKV() *memKV { return &memKV{m: map[string][]byte{}} }
|
||
|
||
func (k *memKV) Get(_ context.Context, key string) ([]byte, bool, error) {
|
||
k.mu.Lock()
|
||
defer k.mu.Unlock()
|
||
v, ok := k.m[key]
|
||
return v, ok, nil
|
||
}
|
||
func (k *memKV) Put(_ context.Context, key string, val []byte) error {
|
||
k.mu.Lock()
|
||
defer k.mu.Unlock()
|
||
k.m[key] = append([]byte(nil), val...)
|
||
return nil
|
||
}
|
||
func (k *memKV) Delete(_ context.Context, key string) error {
|
||
k.mu.Lock()
|
||
defer k.mu.Unlock()
|
||
delete(k.m, key)
|
||
return nil
|
||
}
|
||
|
||
// bus.KVHandle（JetStream KV 后端）须结构化满足 CheckpointKV——钉死 bus 与 eino 的隐式契约，
|
||
// 任一侧改了 Get/Put/Delete 签名都会在这里编译失败。
|
||
var _ CheckpointKV = (*bus.KVHandle)(nil)
|
||
|
||
// TestCheckpointStoreRoundtrip 验证 compose.CheckPointStore 适配：Set→Get 命中、Delete 后 Get 落空。
|
||
func TestCheckpointStoreRoundtrip(t *testing.T) {
|
||
s := newCheckpointStore(newMemKV())
|
||
ctx := context.Background()
|
||
const id = "task_42"
|
||
|
||
if _, ok, err := s.Get(ctx, id); err != nil || ok {
|
||
t.Fatalf("空 store 应未命中: ok=%v err=%v", ok, err)
|
||
}
|
||
|
||
cp := []byte("compose-serialized-state")
|
||
if err := s.Set(ctx, id, cp); err != nil {
|
||
t.Fatalf("Set: %v", err)
|
||
}
|
||
got, ok, err := s.Get(ctx, id)
|
||
if err != nil || !ok || string(got) != string(cp) {
|
||
t.Fatalf("Get 应回放原状态: got=%q ok=%v err=%v", got, ok, err)
|
||
}
|
||
|
||
if err := s.Delete(ctx, id); err != nil {
|
||
t.Fatalf("Delete: %v", err)
|
||
}
|
||
if _, ok, _ := s.Get(ctx, id); ok {
|
||
t.Fatalf("Delete 后不应再命中")
|
||
}
|
||
}
|