diff --git a/cmd/sip-go-agent/dispatcher_command.go b/cmd/sip-go-agent/dispatcher_command.go index 3876dd0..8929ce8 100644 --- a/cmd/sip-go-agent/dispatcher_command.go +++ b/cmd/sip-go-agent/dispatcher_command.go @@ -149,7 +149,7 @@ func runDispatcher(ctx context.Context, mode string) (result error) { worker := &dispatcher.Runtime{ Broker: broker, Bootstrap: dispatcher.Bootstrap{ - DispatcherID: settings.DispatcherID, Client: reader, Store: database, ApplySIP: applySIP, VerifySIP: originator.VerifySIP, Logger: slog.Default(), + DispatcherID: settings.DispatcherID, Client: reader, Store: database, ApplySIP: applySIP, SIPGenerationFloor: originator.SIPGenerationFloor, 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.go b/cmd/sip-go-agent/dispatcher_sip.go index af76292..6a9dbea 100644 --- a/cmd/sip-go-agent/dispatcher_sip.go +++ b/cmd/sip-go-agent/dispatcher_sip.go @@ -97,7 +97,7 @@ func runSIPOnlyDispatcher(ctx context.Context, settings config.DispatcherRuntime originator := &dispatcher.ApprovedOriginator{DispatcherID: settings.DispatcherID, Client: agent, Meta: func(callCtx context.Context) (*agentpb.RequestMeta, error) { return coordinator.ApprovedMeta(callCtx, endpoint.AgentID) }} - lane := dispatcher.SIPOnly{DispatcherID: settings.DispatcherID, Store: db, Client: reader, ApplySIP: originator.ApplySIP, VerifySIP: originator.VerifySIP} + lane := dispatcher.SIPOnly{DispatcherID: settings.DispatcherID, Store: db, Client: reader, ApplySIP: originator.ApplySIP, SIPGenerationFloor: originator.SIPGenerationFloor, VerifySIP: originator.VerifySIP} if _, err := broker.DrainControlPredeclared(ctx, broker.ControlQueue(), lane.HandleNotification); err != nil { return fmt.Errorf("drain assigned SIP controls before startup: %w", err) } diff --git a/internal/asterisk/loader.go b/internal/asterisk/loader.go index 268937f..c33423e 100644 --- a/internal/asterisk/loader.go +++ b/internal/asterisk/loader.go @@ -141,11 +141,23 @@ func (l Loader) observe(ctx context.Context, state appliedState) error { // LoadedSIP never trusts a persisted revision alone: it also checks the file // hash and live PJSIP objects after every Agent boot or Dispatcher query. +// LoadedSIP returns the verified managed trunk revisions. An empty set means +// no Agent-applied state file exists; callers must not treat it as call-ready. func (l Loader) LoadedSIP(ctx context.Context) (map[string]int64, error) { config, statePath, err := l.paths() if err != nil { return nil, err } + info, err := os.Lstat(statePath) + if errors.Is(err, os.ErrNotExist) { + return map[string]int64{}, nil + } + if err != nil { + return nil, fmt.Errorf("inspect last verified SIP revision: %w", err) + } + if !info.Mode().IsRegular() { + return nil, errors.New("last verified SIP revision is not a regular file") + } raw, err := os.ReadFile(statePath) if err != nil { return nil, fmt.Errorf("read last verified SIP revision: %w", err) diff --git a/internal/asterisk/loader_test.go b/internal/asterisk/loader_test.go index 70ec9d8..b2def12 100644 --- a/internal/asterisk/loader_test.go +++ b/internal/asterisk/loader_test.go @@ -121,3 +121,15 @@ func TestLoaderFailsClosedWithoutApprovedStaticTransport(t *testing.T) { t.Fatalf("mutated disk before static transport check: %v", err) } } + +func TestLoaderReportsUninitializedSIPGenerationAsEmpty(t *testing.T) { + dir := t.TempDir() + if err := os.WriteFile(filepath.Join(dir, "pjsip.conf"), []byte("[go-sip-udp]\ntype=transport\nprotocol=udp\nbind=0.0.0.0:5060\n#tryinclude go-sip-managed.conf\n"), 0600); err != nil { + t.Fatal(err) + } + loader := Loader{ConfigDir: dir, Asterisk: "/bin/false", LibraryDir: dir} + loaded, err := loader.LoadedSIP(context.Background()) + if err != nil || len(loaded) != 0 { + t.Fatalf("missing persisted generation must be an explicit empty observation: loaded=%v err=%v", loaded, err) + } +} diff --git a/internal/dispatcher/approved_originator.go b/internal/dispatcher/approved_originator.go index 9896584..93c4dfb 100644 --- a/internal/dispatcher/approved_originator.go +++ b/internal/dispatcher/approved_originator.go @@ -48,7 +48,7 @@ func (o *ApprovedOriginator) activeMeta(ctx context.Context) (*agentpb.RequestMe return proto.Clone(meta).(*agentpb.RequestMeta), nil } -func (o *ApprovedOriginator) LoadedTrunks(ctx context.Context) (map[string]int64, error) { +func (o *ApprovedOriginator) observedLoadedSIP(ctx context.Context) (map[string]int64, error) { meta, err := o.activeMeta(ctx) if err != nil { return nil, err @@ -57,8 +57,8 @@ func (o *ApprovedOriginator) LoadedTrunks(ctx context.Context) (map[string]int64 if err != nil { return nil, fmt.Errorf("inspect Agent SIP load state: %w", err) } - if response == nil || len(response.TrunkRevision) == 0 { - return nil, errors.New("Agent did not report loaded SIP trunks") + if response == nil { + return nil, errors.New("Agent did not report SIP load state") } loaded := make(map[string]int64, len(response.TrunkRevision)) for trunk, revision := range response.TrunkRevision { @@ -70,6 +70,33 @@ func (o *ApprovedOriginator) LoadedTrunks(ctx context.Context) (map[string]int64 return loaded, nil } +// SIPGenerationFloor returns the highest authenticated revision currently +// loaded by Agent. An empty set is only a zero-generation floor, never call authorization. +func (o *ApprovedOriginator) SIPGenerationFloor(ctx context.Context) (int64, error) { + loaded, err := o.observedLoadedSIP(ctx) + if err != nil { + return 0, err + } + var floor int64 + for _, revision := range loaded { + if revision > floor { + floor = revision + } + } + return floor, nil +} + +func (o *ApprovedOriginator) LoadedTrunks(ctx context.Context) (map[string]int64, error) { + loaded, err := o.observedLoadedSIP(ctx) + if err != nil { + return nil, err + } + if len(loaded) == 0 { + return nil, errors.New("Agent did not report loaded SIP trunks") + } + return loaded, nil +} + // ApplySIP sends the complete approved partition through the pinned, active // Agent session. The response must describe the exact revision and trunk set. func (o *ApprovedOriginator) ApplySIP(ctx context.Context, sip configread.SIP) error { diff --git a/internal/dispatcher/approved_originator_test.go b/internal/dispatcher/approved_originator_test.go index a85bbfd..cd8245d 100644 --- a/internal/dispatcher/approved_originator_test.go +++ b/internal/dispatcher/approved_originator_test.go @@ -86,6 +86,27 @@ func approvedOriginatorFixture(t *testing.T, client *fakeApprovedAgentRPC) (*App return orig, spec } +func TestApprovedOriginatorReportsLoadedSIPGenerationFloor(t *testing.T) { + fake := &fakeApprovedAgentRPC{loaded: map[string]int64{"trunk-a": 25, "trunk-b": 26}} + orig, _ := approvedOriginatorFixture(t, fake) + floor, err := orig.SIPGenerationFloor(context.Background()) + if err != nil || floor != 26 { + t.Fatalf("highest authenticated Agent generation not returned: floor=%d err=%v", floor, err) + } + fake.loaded = map[string]int64{} + floor, err = orig.SIPGenerationFloor(context.Background()) + if err != nil || floor != 0 { + t.Fatalf("empty first-load state must report generation floor zero: floor=%d err=%v", floor, err) + } + if _, err := orig.LoadedTrunks(context.Background()); err == nil { + t.Fatal("empty observed set must not authorize a loaded trunk") + } + fake.loadedErr = errors.New("inspection unavailable") + if _, err := orig.SIPGenerationFloor(context.Background()); err == nil { + t.Fatal("Agent inspection failure must not silently use generation zero") + } +} + func TestApprovedOriginatorAppliesFullSIPSnapshotBeforeReportingLoaded(t *testing.T) { fake := &fakeApprovedAgentRPC{loaded: map[string]int64{"trunk-mock": 8}} orig, spec := approvedOriginatorFixture(t, fake) diff --git a/internal/dispatcher/config.go b/internal/dispatcher/config.go index 2c105c3..727fffd 100644 --- a/internal/dispatcher/config.go +++ b/internal/dispatcher/config.go @@ -14,21 +14,39 @@ import ( // configuration contract. Control consumption remains available even when // opening task admission fails. type Bootstrap struct { - Client *configread.Client - Store *store.Store - DispatcherID string - 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 - Providers *map[string]configread.Provider // boot-approved catalog shared by all tasks + Client *configread.Client + Store *store.Store + DispatcherID string + ApplySIP func(context.Context, configread.SIP) error // nil only for isolated Mock; real runtimes apply before verifying + SIPGenerationFloor func(context.Context) (int64, error) // authenticated Agent observation; required when ApplySIP is set + 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 + Providers *map[string]configread.Provider // boot-approved catalog shared by all tasks +} + +func observedSIPGeneration(ctx context.Context, apply bool, observe func(context.Context) (int64, error)) (int64, error) { + if !apply { + return 0, nil + } + if observe == nil { + return 0, errors.New("native SIP apply requires an Agent generation observation") + } + generation, err := observe(ctx) + if err != nil { + return 0, fmt.Errorf("observe Agent SIP generation: %w", err) + } + if generation < 0 { + return 0, errors.New("Agent reported a negative SIP generation") + } + return generation, nil } func (b Bootstrap) Run(ctx context.Context) error { - if b.Client == nil || b.Store == nil || b.DispatcherID == "" || b.VerifySIP == nil || b.DrainControls == nil { - return errors.New("current Dispatcher bootstrap requires config client, durable store, identity, applied SIP verifier, and control drain") + if b.Client == nil || b.Store == nil || b.DispatcherID == "" || b.VerifySIP == nil || b.DrainControls == nil || (b.ApplySIP != nil && b.SIPGenerationFloor == nil) { + return errors.New("current Dispatcher bootstrap requires config client, durable store, identity, SIP generation observation and verifier, and control drain") } if b.Cursor != nil { *b.Cursor = "" @@ -49,7 +67,11 @@ func (b Bootstrap) Run(ctx context.Context) error { if sip.DispatcherID != b.DispatcherID { return errors.New("SIP snapshot belongs to another Dispatcher") } - sip, err = b.Store.PrepareSIP(sip) + generationFloor, err := observedSIPGeneration(ctx, b.ApplySIP != nil, b.SIPGenerationFloor) + if err != nil { + return err + } + sip, err = b.Store.PrepareSIP(sip, generationFloor) if err != nil { return fmt.Errorf("prepare local SIP generation: %w", err) } @@ -62,7 +84,7 @@ func (b Bootstrap) Run(ctx context.Context) error { 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) + b.Logger.Info("startup SIP snapshot prepared", "dispatcher_id", b.DispatcherID, "revision", sip.Revision, "applied_revision", applied, "agent_generation_floor", generationFloor, "occupied_calls", occupied, "native_apply", b.ApplySIP != nil) } if occupied != 0 && applied != sip.Revision { return store.ErrSIPPending diff --git a/internal/dispatcher/config_test.go b/internal/dispatcher/config_test.go index 2ff0bbc..aaf7a6a 100644 --- a/internal/dispatcher/config_test.go +++ b/internal/dispatcher/config_test.go @@ -88,7 +88,7 @@ func TestBootstrapRequiresSIPLoadingAndControlDrain(t *testing.T) { Logger: slog.New(slog.NewJSONHandler(&trace, nil)), ApplySIP: func(_ context.Context, sip configread.SIP) error { stages = append(stages, "apply") - applied = sip.Revision == 8 + applied = sip.Revision == 27 return nil }, VerifySIP: func(_ context.Context, sip configread.SIP) error { @@ -97,11 +97,15 @@ func TestBootstrapRequiresSIPLoadingAndControlDrain(t *testing.T) { if !applied { t.Fatal("verified Agent load before applying SIP") } - if sip.Revision != 8 { + if sip.Revision != 27 { t.Fatalf("verified wrong SIP revision %d", sip.Revision) } return nil }, + SIPGenerationFloor: func(context.Context) (int64, error) { + stages = append(stages, "generation-floor") + return 26, nil + }, DrainControls: func(context.Context) error { drained = true if admitted, err := db.CanAdmit(id, 1001, "task-asr"); err != nil || admitted { @@ -114,10 +118,10 @@ func TestBootstrapRequiresSIPLoadingAndControlDrain(t *testing.T) { t.Fatal(err) } admitted, err := db.CanAdmit(id, 1001, "task-asr") - if err != nil || !admitted || !verifierCalled || !drained || cursor != "opaque-end-token" || approvedSIP.Revision != 8 || len(called) < 6 || called[0] != "/internal/v1/dispatcher/sip" { + if err != nil || !admitted || !verifierCalled || !drained || cursor != "opaque-end-token" || approvedSIP.Revision != 27 || 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"}) { + if !reflect.DeepEqual(stages, []string{"generation-floor", "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"} { @@ -137,11 +141,11 @@ func TestBootstrapRequiresSIPLoadingAndControlDrain(t *testing.T) { t.Fatalf("paused detail opened admission: %v %v", admitted, err) } currentTaskStatus = "running" - bootstrap.DrainControls = func(context.Context) error { return db.NoteSIPChange(id, 9) } + bootstrap.DrainControls = func(context.Context) error { return db.NoteSIPChange(id, 28) } if err := bootstrap.Run(context.Background()); err != nil { t.Fatalf("notification delivered during startup must remain pending for runtime: %v", err) } - if cursor != "opaque-end-token" || approvedSIP.Revision != 8 { + if cursor != "opaque-end-token" || approvedSIP.Revision != 27 { t.Fatalf("pending SIP startup lost verified snapshot/cursor: revision=%d cursor=%q", approvedSIP.Revision, cursor) } if admitted, err := db.CanAdmit(id, 1001, "task-asr"); err != nil || admitted { @@ -194,6 +198,7 @@ func TestBootstrapFailsClosedOnVerifierOrDrainError(t *testing.T) { } return nil }, + SIPGenerationFloor: func(context.Context) (int64, error) { return 0, nil }, VerifySIP: func(context.Context, configread.SIP) error { if failure == "verify" { return errors.New("applied SIP revision unavailable") diff --git a/internal/dispatcher/sip_load_occupancy_test.go b/internal/dispatcher/sip_load_occupancy_test.go index 24f99da..07da1b2 100644 --- a/internal/dispatcher/sip_load_occupancy_test.go +++ b/internal/dispatcher/sip_load_occupancy_test.go @@ -56,9 +56,10 @@ func TestStartupAndSIPNotificationNeverLoadOverOccupiedOrUnknownCalls(t *testing 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 }, + ApplySIP: func(context.Context, configread.SIP) error { applied++; return nil }, + SIPGenerationFloor: func(context.Context) (int64, error) { return 0, 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()) diff --git a/internal/dispatcher/sip_only.go b/internal/dispatcher/sip_only.go index 5fcc8d6..59438b2 100644 --- a/internal/dispatcher/sip_only.go +++ b/internal/dispatcher/sip_only.go @@ -15,11 +15,12 @@ import ( // SIPOnly is a distinct update lane: it never discovers tasks or enables // call admission, even when native Asterisk confirms the entire snapshot. type SIPOnly struct { - DispatcherID string - Store *store.Store - Client *configread.Client - ApplySIP func(context.Context, configread.SIP) error - VerifySIP func(context.Context, configread.SIP) error + DispatcherID string + Store *store.Store + Client *configread.Client + ApplySIP func(context.Context, configread.SIP) error + SIPGenerationFloor func(context.Context) (int64, error) + VerifySIP func(context.Context, configread.SIP) error } // HandleNotification persists the assigned SIP revision before the broker ACK. @@ -55,7 +56,7 @@ func (s SIPOnly) HandleNotification(_ context.Context, route string, body []byte // AI config. The Agent revision and actual native load must both agree before // the durable pending revision is cleared, while admission remains closed. func (s SIPOnly) Sync(ctx context.Context) error { - if s.Store == nil || s.Client == nil || s.DispatcherID == "" || s.ApplySIP == nil || s.VerifySIP == nil { + if s.Store == nil || s.Client == nil || s.DispatcherID == "" || s.ApplySIP == nil || s.SIPGenerationFloor == nil || s.VerifySIP == nil { return errors.New("SIP-only sync requires a bound client, durable store and native Agent") } sip, err := s.Client.ReadSIP(ctx) @@ -65,7 +66,11 @@ func (s SIPOnly) Sync(ctx context.Context) error { if sip.DispatcherID != s.DispatcherID { return errors.New("SIP-only snapshot owner is invalid") } - sip, err = s.Store.PrepareSIP(sip) + generationFloor, err := observedSIPGeneration(ctx, true, s.SIPGenerationFloor) + if err != nil { + return err + } + sip, err = s.Store.PrepareSIP(sip, generationFloor) if err != nil { return err } diff --git a/internal/dispatcher/sip_only_test.go b/internal/dispatcher/sip_only_test.go index e547d87..5d83c80 100644 --- a/internal/dispatcher/sip_only_test.go +++ b/internal/dispatcher/sip_only_test.go @@ -34,9 +34,10 @@ func TestSIPOnlyNotificationCheckpointsNativeLoadWithoutOpeningBusiness(t *testi defer db.Close() applied := false syncer := SIPOnly{DispatcherID: id, Store: db, Client: client, + SIPGenerationFloor: func(context.Context) (int64, error) { return 26, nil }, ApplySIP: func(_ context.Context, sip configread.SIP) error { applied = true - if sip.Revision != 8 { + if sip.Revision != 27 { t.Fatal("unexpected SIP revision") } return nil @@ -58,7 +59,7 @@ func TestSIPOnlyNotificationCheckpointsNativeLoadWithoutOpeningBusiness(t *testi if err := syncer.Sync(context.Background()); err != nil { t.Fatal(err) } - if appliedRev, pending, err := db.SIPState(id); err != nil || appliedRev != 8 || pending != 0 { + if appliedRev, pending, err := db.SIPState(id); err != nil || appliedRev != 27 || pending != 0 { t.Fatalf("verified readback was not recorded: %d %d %v", appliedRev, pending, err) } if err := syncer.HandleNotification(context.Background(), "d."+id+".control.in", configExample(t, "mq-control")); err == nil { diff --git a/internal/dispatcher/sip_reload.go b/internal/dispatcher/sip_reload.go index 8c547cd..05f78a4 100644 --- a/internal/dispatcher/sip_reload.go +++ b/internal/dispatcher/sip_reload.go @@ -37,12 +37,16 @@ func (r *Runtime) refreshSIP(ctx context.Context, approvedSIP *configread.SIP) e if err != nil { return r.closeSIPAdmission(r.rejectSIPSources(ctx, sources, fmt.Errorf("read approved SIP full snapshot: %w", err))) } - current, err = r.Bootstrap.Store.PrepareSIP(current) + generationFloor, err := observedSIPGeneration(ctx, r.Bootstrap.ApplySIP != nil, r.Bootstrap.SIPGenerationFloor) + if err != nil { + return r.closeSIPAdmission(fmt.Errorf("observe Agent SIP generation before reload: %w", err)) + } + current, err = r.Bootstrap.Store.PrepareSIP(current, generationFloor) if err != nil { 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) + r.Logger.Info("applying pending SIP snapshot after calls drained", "dispatcher_id", r.Bootstrap.DispatcherID, "revision", current.Revision, "agent_generation_floor", generationFloor) 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 diff --git a/internal/dispatcher/sip_runtime_test.go b/internal/dispatcher/sip_runtime_test.go index c726570..c4b4608 100644 --- a/internal/dispatcher/sip_runtime_test.go +++ b/internal/dispatcher/sip_runtime_test.go @@ -47,14 +47,15 @@ func TestSIPNotificationPersistsBarrierAndWaitsForLoadedFullSnapshot(t *testing. if applied.Load() == 0 { return errors.New("SIP load checked before native apply") } - if sip.Revision != 9 || !loaded.Load() { + if sip.Revision != 27 || !loaded.Load() { return errors.New("Agent and Asterisk have not applied SIP revision 9") } return nil } runtime := &Runtime{Bootstrap: Bootstrap{DispatcherID: executor.DispatcherID, Client: client, Store: s, + SIPGenerationFloor: func(context.Context) (int64, error) { return 26, nil }, ApplySIP: func(_ context.Context, sip configread.SIP) error { - if sip.Revision != 9 { + if sip.Revision != 27 { return errors.New("wrong SIP revision sent to Agent") } applied.Add(1) @@ -90,11 +91,11 @@ func TestSIPNotificationPersistsBarrierAndWaitsForLoadedFullSnapshot(t *testing. if admitted, err := s.CanAdmit(executor.DispatcherID, 1001, "task-asr"); err != nil || !admitted { t.Fatalf("verified SIP did not reopen: %v %v", admitted, err) } - if approved.Revision != 9 { + if approved.Revision != 27 { t.Fatalf("reloaded SIP not bound: rev=%d", approved.Revision) } bound, err := s.ReadSnapshot(executor.DispatcherID, 1001, "task-asr") - if err != nil || bound.SIP.Revision != 9 { + if err != nil || bound.SIP.Revision != 27 { t.Fatalf("existing task did not inherit the approved SIP revision: %+v %v", bound.SIP, err) } } diff --git a/internal/rpc/approved_execution.go b/internal/rpc/approved_execution.go index 3f4545d..becacce 100644 --- a/internal/rpc/approved_execution.go +++ b/internal/rpc/approved_execution.go @@ -53,7 +53,7 @@ func (s *Server) GetLoadedSIP(ctx context.Context, req *agentpb.GetLoadedSIPRequ return nil, status.Error(codes.FailedPrecondition, "observed SIP load state is unavailable") } loaded, err := s.loadedSIP(ctx) - if err != nil || len(loaded) == 0 { + if err != nil || loaded == nil { return nil, status.Error(codes.Unavailable, "observed SIP load state failed") } copyOfLoaded := make(map[string]int64, len(loaded)) diff --git a/internal/rpc/approved_execution_test.go b/internal/rpc/approved_execution_test.go index e726fea..936e905 100644 --- a/internal/rpc/approved_execution_test.go +++ b/internal/rpc/approved_execution_test.go @@ -149,9 +149,10 @@ func TestExecuteApprovedRejectsMismatchedIdentityAndCredentialsBeforeAttempt(t * func TestGetLoadedSIPRequiresActiveSessionAndReportsActualRevision(t *testing.T) { now := time.Date(2026, 9, 20, 10, 0, 0, 0, time.UTC) req := approvedTestRequest(t, now) + loaded := map[string]int64{req.SelectedTrunkId: 8} s := NewServer(ServerOptions{Mode: "mock", StatePath: filepath.Join(t.TempDir(), "agent-session.json"), Now: func() time.Time { return now }, Status: &agentpb.AgentStatus{AgentId: "agent-1", CellId: "cell-1", BootId: "boot-1"}, - LoadedSIP: func(context.Context) (map[string]int64, error) { return map[string]int64{req.SelectedTrunkId: 8}, nil }, + LoadedSIP: func(context.Context) (map[string]int64, error) { return loaded, nil }, }) if _, err := s.GetLoadedSIP(context.Background(), &agentpb.GetLoadedSIPRequest{Meta: req.Meta}); err == nil { t.Fatal("unactivated caller must not inspect loaded SIP") @@ -166,4 +167,9 @@ func TestGetLoadedSIPRequiresActiveSessionAndReportsActualRevision(t *testing.T) if err != nil || result.TrunkRevision[req.SelectedTrunkId] != 8 { t.Fatalf("must report live loaded revision: %v %v", result, err) } + loaded = map[string]int64{} + result, err = s.GetLoadedSIP(context.Background(), &agentpb.GetLoadedSIPRequest{Meta: req.Meta}) + if err != nil || result == nil || len(result.TrunkRevision) != 0 { + t.Fatalf("must report an explicitly empty state before first SIP load: %v %v", result, err) + } } diff --git a/internal/store/sip_snapshot.go b/internal/store/sip_snapshot.go index f1a1f2f..86e70ea 100644 --- a/internal/store/sip_snapshot.go +++ b/internal/store/sip_snapshot.go @@ -12,10 +12,13 @@ import ( "git.ipao.vip/rogee/go-sip/internal/contract" ) -// PrepareSIP assigns a durable LOCAL load generation. SaaS revision is optional -// and never rejects a newer fetched snapshot. Identical retries reuse the same -// pending generation; changed content gets a new generation, including restart. -func (s *Store) PrepareSIP(sip configread.SIP) (configread.SIP, error) { +// PrepareSIP assigns a durable local load generation. SaaS revision is optional +// and never rejects a fetched snapshot. The observed Agent generation is a +// monotonic floor when SQLite state is missing or behind it. +func (s *Store) PrepareSIP(sip configread.SIP, agentGenerationFloor int64) (configread.SIP, error) { + if agentGenerationFloor < 0 { + return sip, errors.New("Agent SIP generation floor is negative") + } if sip.DispatcherID == "" { return sip, errors.New("SIP snapshot has no owner") } @@ -86,6 +89,12 @@ func (s *Store) PrepareSIP(sip configread.SIP) (configread.SIP, error) { generation++ } generation = max(generation, pending) + if agentGenerationFloor > applied && generation <= agentGenerationFloor { + if agentGenerationFloor == math.MaxInt64 { + return sip, errors.New("local SIP generation cannot advance beyond Agent's loaded generation") + } + generation = agentGenerationFloor + 1 + } if _, err = tx.Exec(`INSERT INTO dispatcher_sip_snapshots(dispatcher_id,revision,content_json) VALUES(?,?,?) ON CONFLICT(dispatcher_id) DO UPDATE SET revision=excluded.revision,content_json=excluded.content_json`, sip.DispatcherID, generation, content); err != nil { return sip, fmt.Errorf("persist local SIP snapshot: %w", err) } diff --git a/internal/store/sip_snapshot_restart_test.go b/internal/store/sip_snapshot_restart_test.go index e28999f..7aa1d6f 100644 --- a/internal/store/sip_snapshot_restart_test.go +++ b/internal/store/sip_snapshot_restart_test.go @@ -23,7 +23,7 @@ func TestLocalSIPPendingContentSurvivesRestartAndForeignOwnerIsRejected(t *testi if err = json.Unmarshal(raw, &source); err != nil { t.Fatal(err) } - first, err := s.PrepareSIP(source) + first, err := s.PrepareSIP(source, 0) if err != nil || first.Revision != 1 { t.Fatalf("first generation: %+v %v", first, err) } @@ -35,13 +35,13 @@ func TestLocalSIPPendingContentSurvivesRestartAndForeignOwnerIsRejected(t *testi t.Fatal(err) } defer s.Close() - same, err := s.PrepareSIP(source) + same, err := s.PrepareSIP(source, 0) if err != nil || same.Revision != 1 { t.Fatalf("restart changed pending generation: %+v %v", same, err) } source.Trunks = json.RawMessage(`[] `) // Encoding whitespace is normalized and is not different content. - same, err = s.PrepareSIP(source) + same, err = s.PrepareSIP(source, 0) if err != nil || same.Revision != 1 { t.Fatalf("equivalent snapshot changed generation: %+v %v", same, err) } @@ -54,16 +54,47 @@ func TestLocalSIPPendingContentSurvivesRestartAndForeignOwnerIsRejected(t *testi t.Fatal(err) } source.Trunks = full.Trunks - changed, err := s.PrepareSIP(source) + changed, err := s.PrepareSIP(source, 0) if err != nil || changed.Revision != 2 { t.Fatalf("pending changed content not assigned new generation: %+v %v", changed, err) } foreign := source foreign.DispatcherID = "4f3aac93-1b25-46b7-a9d7-83d29f5b86e1" - if _, err = s.PrepareSIP(foreign); err == nil { + if _, err = s.PrepareSIP(foreign, 0); err == nil { t.Fatal("foreign owner reused store") } if _, pending, err := s.SIPState(source.DispatcherID); err != nil || pending != 2 { t.Fatalf("foreign request modified pending state: %d %v", pending, err) } } + +func TestPrepareSIPAdvancesBeyondAgentGenerationAfterLocalStateLoss(t *testing.T) { + path := filepath.Join(t.TempDir(), "dispatcher.db") + s, err := Open(path) + if err != nil { + t.Fatal(err) + } + defer s.Close() + raw, err := os.ReadFile("../../contracts/schema/examples/config-read-sip-no-revision.json") + if err != nil { + t.Fatal(err) + } + var source configread.SIP + if err := json.Unmarshal(raw, &source); err != nil { + t.Fatal(err) + } + source.Revision = 25 + + prepared, err := s.PrepareSIP(source, 26) + if err != nil || prepared.Revision != 27 { + t.Fatalf("fresh local state must advance beyond Agent's observed generation: got=%d err=%v", prepared.Revision, err) + } + applied, pending, err := s.SIPState(source.DispatcherID) + if err != nil || applied != 0 || pending != 27 { + t.Fatalf("new generation was not durably fenced: applied=%d pending=%d err=%v", applied, pending, err) + } + retry, err := s.PrepareSIP(source, 26) + if err != nil || retry.Revision != 27 { + t.Fatalf("retry must keep the new generation stable: got=%d err=%v", retry.Revision, err) + } +} diff --git a/internal/store/sip_snapshot_test.go b/internal/store/sip_snapshot_test.go index c510b20..46c802e 100644 --- a/internal/store/sip_snapshot_test.go +++ b/internal/store/sip_snapshot_test.go @@ -14,7 +14,7 @@ func TestSIPMissingOrLowerSourceRevisionUsesDurableLocalGeneration(t *testing.T) } sip := snap.SIP sip.Revision = 0 - first, err := s.PrepareSIP(sip) + first, err := s.PrepareSIP(sip, 0) if err != nil { t.Fatal(err) } @@ -23,20 +23,20 @@ func TestSIPMissingOrLowerSourceRevisionUsesDurableLocalGeneration(t *testing.T) } sip.Trunks = []byte(strings.Replace(string(sip.Trunks), `"dial_prefix":""`, `"dial_prefix":"99"`, 1)) sip.Revision = 1 - next, err := s.PrepareSIP(sip) + next, err := s.PrepareSIP(sip, 0) if err != nil { t.Fatal(err) } if next.Revision != 9 || !strings.Contains(string(next.Trunks), `"dial_prefix":"99"`) { t.Fatalf("lower source revision blocked latest content: %+v", next) } - retry, err := s.PrepareSIP(sip) + retry, err := s.PrepareSIP(sip, 0) if err != nil || retry.Revision != 9 { t.Fatalf("retry changed pending generation: %+v %v", retry, err) } sip.Trunks = []byte(strings.Replace(string(sip.Trunks), `"dial_prefix":"99"`, `"dial_prefix":"88"`, 1)) sip.Revision = 0 - newest, err := s.PrepareSIP(sip) + newest, err := s.PrepareSIP(sip, 0) if err != nil || newest.Revision != 10 { t.Fatalf("new pending content must supersede prior content: %+v %v", newest, err) } @@ -46,7 +46,7 @@ func TestSIPMissingOrLowerSourceRevisionUsesDurableLocalGeneration(t *testing.T) if admitted, err := s.CanAdmit(currentDispatcherID, 1001, "task-asr"); err != nil || admitted { t.Fatalf("unverified SIP reopened admission: %v %v", admitted, err) } - if _, err = s.PrepareSIP(configread.SIP{}); err == nil { + if _, err = s.PrepareSIP(configread.SIP{}, 0); err == nil { t.Fatal("missing owner accepted") } }