diff --git a/AGENTS.md b/AGENTS.md index af13e64..c82281a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -104,7 +104,7 @@ - 呼叫、控制、必要回执和**每通话唯一最终结果**经固定 `v1` RabbitMQ Topic/队列,任务与控制队列由 SaaS 预建,Dispatcher 不可自行建/删/绑定;stop 时先停该任务消费者并关闭持久准入,再 purge 仅该任务队列的待投递消息,失败不回成功;SaaS 停止继续投递已 stop 任务,后续误投递不执行。结果进入指定共享 durable 队列。独立 D UUID 和接收队列不能广播后正文过滤;`tenant_key` 原值保留,任务只能由归属 D 执行。入站先校验和持久 inbox 后 ACK,状态/outbox 同事务;persistent、mandatory、无 return、publisher confirm 成功才记交付,confirm **不是** SaaS 应用收讫。失败/确认丢失与重启只重发同身份消息,不重复拨号或捏造结果。 - `call.execute` 只带获批 `task_id/callee`,调用线路、AI 和时限由该任务快照固定;主叫由绑定线路的 SIP 快照唯一 `caller_id` 决定,不再由任务挑选主叫档案;`task.control` 的 start/pause/resume/stop/edit 无旧 CAS/版本字段,控制回 task/action/status 处理结果,stop 不可恢复。任务启动/重启完整读取一次列表,运行中不定时轮询任务列表或任务配置;新建任务由 start 读取任务配置和租户额度,resume 重读并解除人工暂停,edit 只重读、校验并持久更新已有任务的配置和额度,不改变人工暂停/停止状态、不向 Agent 发动作,也不影响已发出的通话快照;旧修订号及停止后 edit 拒绝。全局 AI 服务商列表每次启动只读取一次,本进程全部任务复用,变更须重启后才生效。启动时全量读取并核验 SIP,运行中只由 `sip.config` 通知触发全量 SIP 读取,无常规定时 SIP 轮询;任务 start/resume/edit 使用已生效 SIP 快照,不自行拉取或向 Agent 核验 SIP。SIP 修订变化待旧呼叫结束且 Agent 已加载后,仅在本地重新绑定任务快照,不重读任务列表;待处理期间新准入关闭。发现分页同快照先全部校验后提交,MQ 控制持久状态高于偶发 HTTP 状态;冲突关准入、resume 须重新核验。历史命令不因消息年龄过期,但实际呼出前仍检查任务归属/事件号码格式/时段/授权/额度和有效通话上限;任务结束不删未确认结果、恢复、幂等或未知占用。 - 独立 Dispatcher 的 SQLite 是任务、额度、inbox/outbox 的权威数据;Agent 无业务数据库,录音、执行与上传恢复只写受控私有文件。额度包含未知占用,新 boot/租约到期不得自动清除未知执行;明确 Agent 在拨号前拒绝时须持久拒绝并释放,已接纳但证实 ARI originate 从未提交的失败由原 Agent 经正式终结回执释放,RPC 交付未知时仍占用。2026-10-04 修复版 `d2c4a99` 已安装并通过非呼出时段主机诊断,新隔离测试租户额度 2/revision 2 已启用并由 Dispatcher 只读拉取;原快照保留;该条旧执行仅按前述经使用者授权的人工核实记录解除 Dispatcher 占用,绝非新 boot 或超时自动清除未知执行。仅 Agent 用机器可读标记证实本次未发起的拒绝可自动释放;旧执行的同错误码不算证明。不实现双活数据库、自动跨机热备、多 D 共享额度或第二租户公平。本轮不借本机 D1/D2 隔离夹具宣称多 D 运行。不得建立旧表/旧消息/旧 HTTP 执行兼容通道。 -- Dispatcher↔Agent 复用 Unary gRPC 和受控 Endpoint;Agent 预绑定 D UUID 与服务端证书指纹,激活/会话代际、peer mTLS/SAN/SNI 和已签发期限须核对,新 boot 不清未知占用。Agent 不自行向 SaaS 取任务/AI/OSS 授权;Dispatcher 只用已经核验的 Agent `GetLoadedSIP` revision 开执行准入。本机 Mock 的加载回报不证明 Asterisk 已实际加载;仅隔离 SIP-only Agent 在配置原子写入、PJSIP reload 和运行态 endpoint/AOR/UDP transport 一致后持久标记 revision,每次加载查询重新核验。只支持明确的 UDP、IP/none 鉴权、无需 REGISTER 的 IPv4/PCMA 线路;未知字段和其他传输/鉴权/注册方式拒绝,不热更静态 transport。每条 SIP trunk 只允许一个原值 `caller_id`,Agent 把该值同时配置为 Asterisk endpoint 的 `from_user` 和 `callerid`,并在 `GetLoadedSIP` 前核对实际运行值;旧任务 `caller_profile_id` 与线路 `caller_profiles` 必须直接拒绝,不保留兼容路径。SIP 配置的唯一编辑/审批面仍是 management。 +- Dispatcher↔Agent 复用 Unary gRPC 和受控 Endpoint;Agent 预绑定 D UUID 与服务端证书指纹,激活/会话代际、peer mTLS/SAN/SNI 和已签发期限须核对,新 boot 不清未知占用。Agent 不自行向 SaaS 取任务/AI/OSS 授权;Dispatcher 只用已经核验的 Agent `GetLoadedSIP` revision 开执行准入。本机 Mock 的加载回报不证明 Asterisk 已实际加载;真实 Agent(`nonprod-real` 与隔离 `sip-only`)在配置原子写入、PJSIP reload 和运行态 endpoint/AOR/UDP transport 一致后持久标记 revision,每次加载查询重新核验。正式业务 Dispatcher 在启动及 SIP 更新时必须通过现有 Agent 会话加载获批快照后再核对版本,不得漏接加载步骤;旧通话与未知占用保留、排空前关新准入。隔离 Mock 仍只核对预加载夹具,不发起原生 SIP reload。只支持明确的 UDP、IP/none 鉴权、无需 REGISTER 的 IPv4/PCMA 线路;未知字段和其他传输/鉴权/注册方式拒绝,不热更静态 transport。每条 SIP trunk 只允许一个原值 `caller_id`,Agent 把该值同时配置为 Asterisk endpoint 的 `from_user` 和 `callerid`,并在 `GetLoadedSIP` 前核对实际运行值;旧任务 `caller_profile_id` 与线路 `caller_profiles` 必须直接拒绝,不保留兼容路径。SIP 配置的唯一编辑/审批面仍是 management。 - 非生产真实 Agent 的 Asterisk `res_hep`/`res_hep_pjsip` 仅镜像至本机 UDP:在 ARI Dial 前读取本次 SIP Call-ID 并绑定执行,按 Call-ID 和原始 INVITE transaction 只采集真实最终响应的状态码、状态行与完整 `raw`;丢失/无法关联以 `sip_capture_error` 显式报告,不从 ARI、号码或挂断原因推测。镜像缺失不阻止已证实终结的额度释放;未经确认的占用仍保留。配置及回滚见 [`deploys/cell/README.md`](deploys/cell/README.md),不得将完整 SIP 报文写入普通日志、提交或聊天;本地测试通过不等于测试机已部署 HEP 或真实接通。 - 2026-10-08 用户批准本地配置规则调整及新模型调用支持:SaaS SIP `revision` 可省略,完整读取覆盖旧配置;`transport/auth_mode` 判定忽略大小写,但不补其它缺失线路参数。Dispatcher 持久分配内部 SIP 加载代次并核对 Agent/Asterisk,旧通话排空与未知占用保留规则不变。任务列表仅忽略 `schema_version/control_seq`。任务用唯一的 `provider_ref` 对应连接的 `provider_code`,`provider_id` 仅为连接记录标识;一份连接可供多用途复用;任务决定厂商/模型/ASR–LLM–TTS 用途,provider 只提供连接信息,不要求 `role/adapter/enabled`,未知供应商/协议及不可表达的连接参数显式拒绝。2026-10-08 追加用户确认:移除精确型号/音色业务硬编码;同协议模型及可表达参数从任务取得。当前 TTS 固定使用 task-based WebSocket,不按 model 名或地址猜协议、不失败回退 HTTP。模型实际可用性由供应商调用返回事实决定,不以本地模拟代签。新增本地调用代码及模拟测试不授权部署、拨号或真实 AI 请求。SQLite 当前布局因新增持久 SIP 快照变为 3;旧布局只读拒绝,不自动迁移或改写旧文件。 - 2026-10-08 用户确认按共享合同 `b502ad2` 本地适配任务线路并发:`allowed_trunk_ids` 只接受包含 `trunk_id/concurrency` 的对象数组,重复线路拒绝;非负整数 `concurrency` 的 0 表示该任务不使用该线路。选线及 SQLite 原子占用同时遵守租户、任务总量、线路总量、任务在线路上的上限,未知执行不释放;任务修订在选线后变化时拒绝原选择,不用旧快照发出新呼叫。占用仍由现有 inbox 与冻结任务快照确定,不新增业务表、不转换旧快照或迁移旧库。ASR/TTS 使用连接的 `ws_endpoint`,LLM 使用 `api_endpoint`,缺失不互相回退。本轮仅本地代码、测试和说明,不部署、不拨号、不请求真实 AI。 diff --git a/cmd/sip-go-agent/dispatcher_command.go b/cmd/sip-go-agent/dispatcher_command.go index d2a84a0..e777ecc 100644 --- a/cmd/sip-go-agent/dispatcher_command.go +++ b/cmd/sip-go-agent/dispatcher_command.go @@ -135,10 +135,14 @@ func runDispatcher(ctx context.Context, mode string) (result error) { return coordinator.ApprovedMeta(callCtx, agentEndpoint.AgentID) }, } + applySIP, err := dispatcherSIPApplier(mode, originator) + if err != nil { + return err + } worker := &dispatcher.Runtime{ Broker: broker, Bootstrap: dispatcher.Bootstrap{ - DispatcherID: settings.DispatcherID, Client: reader, Store: database, VerifySIP: originator.VerifySIP, + DispatcherID: settings.DispatcherID, Client: reader, Store: database, ApplySIP: applySIP, VerifySIP: originator.VerifySIP, Logger: slog.Default(), }, Execute: dispatcher.ExecuteController{ DispatcherID: settings.DispatcherID, Store: database, Originator: originator, Publisher: broker, Now: time.Now, diff --git a/cmd/sip-go-agent/dispatcher_sip_apply.go b/cmd/sip-go-agent/dispatcher_sip_apply.go new file mode 100644 index 0000000..fa6017e --- /dev/null +++ b/cmd/sip-go-agent/dispatcher_sip_apply.go @@ -0,0 +1,26 @@ +package main + +import ( + "context" + "errors" + "fmt" + + "git.ipao.vip/rogee/go-sip/internal/configread" + "git.ipao.vip/rogee/go-sip/internal/dispatcher" +) + +func dispatcherSIPApplier(mode string, originator *dispatcher.ApprovedOriginator) (func(context.Context, configread.SIP) error, error) { + switch mode { + case "nonprod-real": + if originator == nil || originator.DispatcherID == "" || originator.Client == nil || originator.Meta == nil { + return nil, errors.New("real Dispatcher requires an active Agent SIP loading session") + } + return originator.ApplySIP, nil + case "mock": + // Mock's fixed, explicitly preloaded fixture is only verified. It must + // never issue a native reload or pretend that Asterisk loaded it. + return nil, nil + default: + return nil, fmt.Errorf("unsupported business SIP loading mode %q", mode) + } +} diff --git a/cmd/sip-go-agent/dispatcher_sip_apply_test.go b/cmd/sip-go-agent/dispatcher_sip_apply_test.go new file mode 100644 index 0000000..707a63c --- /dev/null +++ b/cmd/sip-go-agent/dispatcher_sip_apply_test.go @@ -0,0 +1,150 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "os" + "reflect" + "testing" + + agentpb "git.ipao.vip/rogee/go-sip/gen/agent" + "git.ipao.vip/rogee/go-sip/internal/configread" + "git.ipao.vip/rogee/go-sip/internal/dispatcher" + "google.golang.org/grpc" +) + +type startupSIPAgent struct { + agentpb.AgentControlServiceClient + loaded map[string]int64 + actions []string + applyErr error + request *agentpb.ApplySIPRequest +} + +func (a *startupSIPAgent) ApplySIP(_ context.Context, request *agentpb.ApplySIPRequest, _ ...grpc.CallOption) (*agentpb.ApplySIPResponse, error) { + a.actions = append(a.actions, "apply") + a.request = request + if a.applyErr != nil { + return nil, a.applyErr + } + var snapshot configread.SIP + if err := json.Unmarshal(request.ApprovedSnapshotJson, &snapshot); err != nil { + return nil, err + } + var trunks []struct { + ID string `json:"trunk_id"` + Enabled bool `json:"enabled"` + } + if err := json.Unmarshal(snapshot.Trunks, &trunks); err != nil { + return nil, err + } + a.loaded = make(map[string]int64) + for _, trunk := range trunks { + if trunk.Enabled { + a.loaded[trunk.ID] = snapshot.Revision + } + } + return &agentpb.ApplySIPResponse{TrunkRevision: a.loaded}, nil +} + +func (a *startupSIPAgent) GetLoadedSIP(context.Context, *agentpb.GetLoadedSIPRequest, ...grpc.CallOption) (*agentpb.GetLoadedSIPResponse, error) { + a.actions = append(a.actions, "verify") + return &agentpb.GetLoadedSIPResponse{TrunkRevision: a.loaded}, nil +} + +func startupSIPFixture(t *testing.T) (*dispatcher.ApprovedOriginator, *startupSIPAgent, configread.SIP) { + t.Helper() + body, err := os.ReadFile("../../contracts/schema/examples/config-read-sip.json") + if err != nil { + t.Fatal(err) + } + var snapshot configread.SIP + if err := json.Unmarshal(body, &snapshot); err != nil { + t.Fatal(err) + } + snapshot.Revision = 16 + var trunks []struct { + ID string `json:"trunk_id"` + Enabled bool `json:"enabled"` + } + if err := json.Unmarshal(snapshot.Trunks, &trunks); err != nil { + t.Fatal(err) + } + agent := &startupSIPAgent{loaded: make(map[string]int64)} + for _, trunk := range trunks { + if trunk.Enabled { + agent.loaded[trunk.ID] = 14 + } + } + originator := &dispatcher.ApprovedOriginator{ + DispatcherID: snapshot.DispatcherID, + Client: agent, + Meta: func(context.Context) (*agentpb.RequestMeta, error) { + return &agentpb.RequestMeta{AgentId: "agent-test", CellId: "cell-test", BootId: "boot-test", DispatcherEpoch: "epoch-test", SessionGeneration: 1}, nil + }, + } + return originator, agent, snapshot +} + +func TestDispatcherRealSIPApplierLoadsOldAgentBeforeVerification(t *testing.T) { + originator, agent, snapshot := startupSIPFixture(t) + if err := originator.VerifySIP(context.Background(), snapshot); err == nil { + t.Fatal("test Agent must start on the old revision") + } + agent.actions = nil + apply, err := dispatcherSIPApplier("nonprod-real", originator) + if err != nil { + t.Fatal(err) + } + if apply == nil { + t.Fatal("real Dispatcher startup omitted the mandatory SIP loading step") + } + if err := apply(context.Background(), snapshot); err != nil { + t.Fatal(err) + } + if err := originator.VerifySIP(context.Background(), snapshot); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(agent.actions, []string{"apply", "verify"}) { + t.Fatalf("SIP lifecycle = %v", agent.actions) + } + if agent.request == nil || agent.request.Meta.IdempotencyKey == "" { + t.Fatal("SIP load lost its full snapshot or operation identity") + } +} + +func TestDispatcherRealSIPApplierPreservesLoadFailure(t *testing.T) { + originator, agent, snapshot := startupSIPFixture(t) + failure := errors.New("native reload failed") + agent.applyErr = failure + apply, err := dispatcherSIPApplier("nonprod-real", originator) + if err != nil { + t.Fatal(err) + } + if apply == nil { + t.Fatal("real Dispatcher has no SIP loader") + } + if err := apply(context.Background(), snapshot); !errors.Is(err, failure) { + t.Fatalf("SIP load error was hidden: %v", err) + } + if err := originator.VerifySIP(context.Background(), snapshot); err == nil { + t.Fatal("failed native load must not be reported as the new revision") + } +} + +func TestDispatcherSIPApplierRequiresRealSessionAndKeepsMockPreloaded(t *testing.T) { + for _, originator := range []*dispatcher.ApprovedOriginator{nil, {DispatcherID: "test"}} { + if _, err := dispatcherSIPApplier("nonprod-real", originator); err == nil { + t.Fatal("real Dispatcher accepted a missing SIP loading session") + } + } + if apply, err := dispatcherSIPApplier("mock", nil); err != nil || apply != nil { + t.Fatal("isolated Mock must only verify its preloaded fixture, not issue native SIP loads") + } + for _, mode := range []string{"", "real", "mixed", "sip-only"} { + if _, err := dispatcherSIPApplier(mode, nil); err == nil { + t.Fatalf("unsupported business mode %q was accepted", mode) + } + } +} diff --git a/deploys/production/README.md b/deploys/production/README.md index 5136a21..ea19a29 100644 --- a/deploys/production/README.md +++ b/deploys/production/README.md @@ -44,6 +44,8 @@ 用户已授权本次目标中的真实 SaaS 全部获批任务,不再逐通询问。程序仍核对任务归属、时段、线路、额度、并发、签发期限和消息身份;不自行生成外呼指令,不盲重拨或换线重拨。 +正式业务 Dispatcher 启动及收到 SIP 更新时,先关闭新执行准入、取得并持久保存 SaaS 完整 SIP 配置,等待旧通话及占用排空,再通过已有 Agent 会话加载并核对 Asterisk 实际版本,成功后才继续业务。未知占用不会因重启消失;加载或核对失败不跳过、不伪造确认。只更新项目管理的线路文件并执行 PJSIP reload,不重启 Asterisk、不改静态 transport。隔离 Mock 仍只核对固定预加载夹具,不发起原生加载。 + 验收包括三项服务安装、启用并启动,Asterisk/Agent `enabled+active`,外部 RabbitMQ 登录/通道、Agent监听、Asterisk基础运行及版本/hash。SaaS 未启动不阻断部署:Dispatcher可明确等待或自动重试,新执行准入保持关闭;SaaS读取、其业务队列供给、业务会话及新SIP加载延期,不把此状态写成业务已就绪。证书、配置、MQ登录及Asterisk故障不得归为SaaS不可用。没有HTTP响应须按真实错误类型判定,证书或未知传输错误阻断部署;等待状态必须是有最新对应错误的 `activating/auto-restart`,`failed` 不算成功。无获批任务、未接通或AI未调用时如实记录未验证。 ## 一键入口 diff --git a/docs/evidence/dispatcher-sip-startup-load-20261009.md b/docs/evidence/dispatcher-sip-startup-load-20261009.md new file mode 100644 index 0000000..bb1cada --- /dev/null +++ b/docs/evidence/dispatcher-sip-startup-load-20261009.md @@ -0,0 +1,38 @@ +# 正式 Dispatcher SIP 加载遗漏修复(2026-10-09) + +## 授权与完成标准 + +使用者在只读根因定位后确认按已讨论方案修复、验证并重新部署 `server.sip`:正式入口接入现有 SIP 加载能力,保留占用检查与实际版本核对,补启动和更新回归测试。 + +不重置、清理或迁移业务数据,不创建或删除真实 RabbitMQ 拓扑,不重启 Asterisk,不做独立试拨或主动投递任务。正式部署仍沿用固定配置和数据目录、`nonprod-real` 与既有真实 SaaS 获批任务授权。 + +完成标准:旧逻辑回归测试明确失败、修复后通过;完整本地验收和业务覆盖率 ≥65%;代码先提交并推送再部署;现场核对实际 SIP 加载与服务状态,不以 Mock 或部署标签替代事实。 + +## 根因及只读现场证据 + +- 正式业务入口只连接了 `VerifySIP`,遗漏 `ApplySIP`。启动和运行中 SIP 更新都使用这组回调,所以读到新配置后只检查它是否已经加载,而不发出加载请求。 +- 15:42 SaaS SIP HTTP 200、归属匹配、三条配置通过项目校验;源 revision 16,数据与 Dispatcher 保存的快照一致。 +- Dispatcher 待加载代次 16,Agent 应用记录及原生 Asterisk 均为 14。15:47 原生查询确认旧三条 endpoint/AOR、主叫、contact、UDP 和 PCMA 与应用记录一致,零活动通道、零执行占用。 +- 因此这次加载失败是本项目入口遗漏,不是当前 SaaS SIP 数据错误;此判断不承诺尚未核验的任务、AI 数据或供应商可接通性。 + +## 改动 + +- 正式业务入口明确按运行模式选择 SIP 加载回调:`nonprod-real` 使用现有 `ApprovedOriginator.ApplySIP`,缺少真实 Agent 会话直接报错,未知模式拒绝。 +- 将该回调接入业务 `Bootstrap`,启动和 `sip.config` 更新共用已有的加载、排空与实际版本核对流程,不新增加载协议、不手工绕过业务流程。 +- 隔离 Mock 保持原先只核对固定预加载夹具的行为,不发原生 reload、不把模拟响应声称为 Asterisk 加载事实。SIP-only 继续使用其独立已有流程。 +- 补启动配置已读取及持久准备、占用数量、实际加载与版本核对的脱敏日志,运行期记录排空后加载;不记录密钥、完整 SIP 或配置内容。 +- 同步 `AGENTS.md` 和正式部署说明中的实际加载边界。 + +## 本地验证 + +- 正式入口加载选择回归:测试 Agent 先报告版本 14,新快照版本 16;旧实现缺少加载回调明确失败。修复后先调用 Apply,再核对新版本;加载失败原错误保留,不伪造新版本。 +- 启动测试确认 Apply → Verify 顺序与三个关键日志节点。 +- 启动/运行通知 × 活动/未知执行四组占用测试均通过:不加载、不提前核对、不开放新准入、不释放占用。 +- 既有运行中 SIP 更新测试确认加载后仍等待真实核对,不成功时维持关闭准入。 +- `make proto-check acceptance-local`、`make release-check-local`、`go vet ./...`、`go test -race ./...`、构建、隔离 RabbitMQ/HTTPS/双向 TLS 集成测试均通过。 +- 业务覆盖率 **70.8%**。 +- 共享契约仍固定在 `4f515796e30c0cfd6f86b39ec2324fa9a8b2b964`,本轮未修改契约。Mock 与本地结果不代表真实通话、AI、OSS 或生产签收。 + +## 部署验证 + +部署后补充固定提交、制品 hash、Agent/Asterisk 实际加载版本、服务状态和剩余阻碍;未核对前不标记部署完成。 diff --git a/internal/dispatcher/config.go b/internal/dispatcher/config.go index 29b4260..c3284fa 100644 --- a/internal/dispatcher/config.go +++ b/internal/dispatcher/config.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "log/slog" "git.ipao.vip/rogee/go-sip/internal/configread" "git.ipao.vip/rogee/go-sip/internal/store" @@ -16,8 +17,9 @@ type Bootstrap struct { Client *configread.Client Store *store.Store DispatcherID string - ApplySIP func(context.Context, configread.SIP) error // nil for isolated Mock; real SIP-only must apply before verifying + ApplySIP func(context.Context, configread.SIP) error // nil only for isolated Mock; real runtimes apply before verifying VerifySIP func(context.Context, configread.SIP) error + Logger *slog.Logger DrainControls func(context.Context) error Cursor *string // memory-only; restart always starts from a full snapshot SIP *configread.SIP // the exact approved snapshot verified on startup @@ -59,10 +61,16 @@ func (b Bootstrap) Run(ctx context.Context) error { if err != nil { return err } + if b.Logger != nil { + b.Logger.Info("startup SIP snapshot prepared", "dispatcher_id", b.DispatcherID, "revision", sip.Revision, "applied_revision", applied, "occupied_calls", occupied, "native_apply", b.ApplySIP != nil) + } if occupied != 0 && applied != sip.Revision { return store.ErrSIPPending } if b.ApplySIP != nil { + if b.Logger != nil { + b.Logger.Info("applying startup SIP snapshot", "dispatcher_id", b.DispatcherID, "revision", sip.Revision) + } if err := b.ApplySIP(ctx, sip); err != nil { return fmt.Errorf("apply approved SIP revision %d on Agent/Asterisk: %w", sip.Revision, err) } @@ -70,6 +78,9 @@ func (b Bootstrap) Run(ctx context.Context) error { if err := b.VerifySIP(ctx, sip); err != nil { return fmt.Errorf("verify SIP revision %d loaded by Agent/Asterisk: %w", sip.Revision, err) } + if b.Logger != nil { + b.Logger.Info("startup SIP snapshot verified", "dispatcher_id", b.DispatcherID, "revision", sip.Revision, "native_apply", b.ApplySIP != nil) + } providers, err := b.Client.ReadProviders(ctx) if err != nil { return fmt.Errorf("read boot-approved AI providers: %w", err) diff --git a/internal/dispatcher/config_test.go b/internal/dispatcher/config_test.go index 59f2298..64724ab 100644 --- a/internal/dispatcher/config_test.go +++ b/internal/dispatcher/config_test.go @@ -1,12 +1,16 @@ package dispatcher import ( + "bytes" "context" "errors" + "log/slog" "net/http" "net/http/httptest" "os" "path/filepath" + "reflect" + "strings" "testing" "git.ipao.vip/rogee/go-sip/internal/configread" @@ -62,15 +66,20 @@ func TestBootstrapRequiresSIPLoadingAndControlDrain(t *testing.T) { } defer db.Close() applied, verifierCalled, drained := false, false, false + var stages []string + var trace bytes.Buffer var cursor string var approvedSIP configread.SIP bootstrap := Bootstrap{ Client: client, Store: db, DispatcherID: id, Cursor: &cursor, SIP: &approvedSIP, + Logger: slog.New(slog.NewJSONHandler(&trace, nil)), ApplySIP: func(_ context.Context, sip configread.SIP) error { + stages = append(stages, "apply") applied = sip.Revision == 8 return nil }, VerifySIP: func(_ context.Context, sip configread.SIP) error { + stages = append(stages, "verify") verifierCalled = true if !applied { t.Fatal("verified Agent load before applying SIP") @@ -95,6 +104,14 @@ func TestBootstrapRequiresSIPLoadingAndControlDrain(t *testing.T) { if err != nil || !admitted || !verifierCalled || !drained || cursor != "opaque-end-token" || approvedSIP.Revision != 8 || len(called) < 6 || called[0] != "/internal/v1/dispatcher/sip" { t.Fatalf("bootstrap: admitted=%v, err=%v, verified=%v, drained=%v, paths=%v", admitted, err, verifierCalled, drained, called) } + if !reflect.DeepEqual(stages, []string{"apply", "verify"}) { + t.Fatalf("startup SIP lifecycle = %v", stages) + } + for _, message := range []string{"startup SIP snapshot prepared", "applying startup SIP snapshot", "startup SIP snapshot verified"} { + if !strings.Contains(trace.String(), message) { + t.Fatalf("missing SIP trace %q", message) + } + } bootstrap.DrainControls = func(context.Context) error { return db.NoteSIPChange(id, 9) } if err := bootstrap.Run(context.Background()); err != nil { t.Fatalf("notification delivered during startup must remain pending for runtime: %v", err) diff --git a/internal/dispatcher/sip_load_occupancy_test.go b/internal/dispatcher/sip_load_occupancy_test.go new file mode 100644 index 0000000..24f99da --- /dev/null +++ b/internal/dispatcher/sip_load_occupancy_test.go @@ -0,0 +1,87 @@ +package dispatcher + +import ( + "context" + "encoding/json" + "errors" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "testing" + + "git.ipao.vip/rogee/go-sip/internal/configread" + "git.ipao.vip/rogee/go-sip/internal/store" +) + +func TestStartupAndSIPNotificationNeverLoadOverOccupiedOrUnknownCalls(t *testing.T) { + for _, outcome := range []string{"active", "unknown"} { + for _, phase := range []string{"startup", "notification"} { + t.Run(outcome+"/"+phase, func(t *testing.T) { + execute, originator, _, database := newExecuteFixture(t) + if outcome == "unknown" { + originator.err = context.DeadlineExceeded + } + err := execute.ProcessExecute(context.Background(), executeBody(t, "sip-load-occupied", "15003164745")) + if (outcome == "unknown") != (err != nil) { + t.Fatalf("unexpected isolated execution state: %v", err) + } + occupied, err := database.OccupiedCalls(execute.DispatcherID) + if err != nil || occupied != 1 { + t.Fatalf("occupied before update = %d, error = %v", occupied, err) + } + var snapshot configread.SIP + if err := json.Unmarshal(configExample(t, "config-read-sip"), &snapshot); err != nil { + t.Fatal(err) + } + snapshot.Revision = 9 + body, err := json.Marshal(snapshot) + if err != nil { + t.Fatal(err) + } + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/internal/v1/dispatcher/sip" { + t.Errorf("occupied startup requested unrelated configuration: %s", r.URL.Path) + w.WriteHeader(http.StatusNotFound) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write(body) + })) + defer server.Close() + client, err := configread.NewClient(server.URL, execute.DispatcherID, "test-secret", server.Client()) + if err != nil { + t.Fatal(err) + } + applied, verified := 0, 0 + bootstrap := Bootstrap{ + Client: client, Store: database, DispatcherID: execute.DispatcherID, + ApplySIP: func(context.Context, configread.SIP) error { applied++; return nil }, + VerifySIP: func(context.Context, configread.SIP) error { verified++; return nil }, + DrainControls: func(context.Context) error { return nil }, + } + if phase == "startup" { + err = bootstrap.Run(context.Background()) + } else { + if err := database.NoteSIPChange(execute.DispatcherID, 9); err != nil { + t.Fatal(err) + } + runtime := &Runtime{Bootstrap: bootstrap, Logger: slog.New(slog.NewTextHandler(io.Discard, nil))} + current := policySnapshot(t).SIP + err = runtime.refreshSIP(context.Background(), ¤t) + } + fenced := phase == "startup" && errors.Is(err, store.ErrSIPPending) || phase == "notification" && err == nil + if !fenced || applied != 0 || verified != 0 { + t.Fatalf("occupied SIP load was not fenced: error=%v applied=%d verified=%d", err, applied, verified) + } + occupied, err = database.OccupiedCalls(execute.DispatcherID) + if err != nil || occupied != 1 { + t.Fatalf("SIP update released occupied execution: %d %v", occupied, err) + } + if admitted, err := database.CanAdmit(execute.DispatcherID, 1001, "task-asr"); err != nil || admitted { + t.Fatalf("occupied SIP update reopened admission: %v %v", admitted, err) + } + }) + } + } +} diff --git a/internal/dispatcher/sip_reload.go b/internal/dispatcher/sip_reload.go index 1d424eb..92042d1 100644 --- a/internal/dispatcher/sip_reload.go +++ b/internal/dispatcher/sip_reload.go @@ -38,6 +38,7 @@ func (r *Runtime) refreshSIP(ctx context.Context, approvedSIP *configread.SIP) e return r.closeSIPAdmission(fmt.Errorf("persist local SIP load generation: %w", err)) } if r.Bootstrap.ApplySIP != nil { + r.Logger.Info("applying pending SIP snapshot after calls drained", "dispatcher_id", r.Bootstrap.DispatcherID, "revision", current.Revision) if err := r.Bootstrap.ApplySIP(ctx, current); err != nil { r.Logger.Warn("approved SIP update remains fenced until native reload confirms", "dispatcher_id", r.Bootstrap.DispatcherID, "revision", current.Revision, "error", err) return nil