rogee/go-sip
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(dispatcher): reconcile SIP generation with Agent state

Browse Source
This commit is contained in:
rogee
2026-10-11 23:01:21 +08:00
parent c3357ab397
commit 16a98414cb
18 changed files with 216 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/sip-go-agent/dispatcher_command.go
+1 -1
View File
@@ -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,
cmd/sip-go-agent/dispatcher_sip.go
+1 -1
View File
@@ -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)
}
internal/asterisk/loader.go
+12
View File
@@ -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)
internal/asterisk/loader_test.go
+12
View File
@@ -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)
}
}
internal/dispatcher/approved_originator.go
+30 -3
View File
@@ -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 {
internal/dispatcher/approved_originator_test.go
+21
View File
@@ -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)
internal/dispatcher/config.go
+36 -14
View File
@@ -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
internal/dispatcher/config_test.go
+11 -6
View File
@@ -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")
internal/dispatcher/sip_load_occupancy_test.go
+4 -3
View File
@@ -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())
internal/dispatcher/sip_only.go
+12 -7
View File
@@ -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
}
internal/dispatcher/sip_only_test.go
+3 -2
View File
@@ -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 {
internal/dispatcher/sip_reload.go
+6 -2
View File
@@ -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
internal/dispatcher/sip_runtime_test.go
+5 -4
View File
@@ -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)
}
}
internal/rpc/approved_execution.go
+1 -1
View File
@@ -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))
internal/rpc/approved_execution_test.go
+7 -1
View File
@@ -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)
}
}
internal/store/sip_snapshot.go
+13 -4
View File
@@ -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)
}
internal/store/sip_snapshot_restart_test.go
+36 -5
View File
@@ -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)
}
}
internal/store/sip_snapshot_test.go
+5 -5
View File
@@ -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")
}
}