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

fix(dispatcher): apply approved SIP before real startup verification

Browse Source
This commit is contained in:
rogee
2026-10-09 16:20:37 +08:00
parent f323e31b6c
commit b37d9f3af0
10 changed files with 339 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/dispatcher/config.go
+12 -1
View File
@@ -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)
internal/dispatcher/config_test.go
+17
View File
@@ -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)
internal/dispatcher/sip_load_occupancy_test.go
+87
View File
@@ -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(), &current)
}
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)
}
})
}
}
}
internal/dispatcher/sip_reload.go
+1
View File
@@ -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