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

Load AI providers once for each Dispatcher run

Browse Source
This commit is contained in:
rogee
2026-10-01 16:22:25 +08:00
parent 40e531cce3
commit a9c8afcab5
7 changed files with 123 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/configread/snapshots.go
+27 -18
View File
@@ -119,36 +119,45 @@ func (c *Client) ReadSIP(ctx context.Context) (SIP, error) {
return sip, nil
}
// ReadTask reads task, provider and quota resources using the already approved
// SIP snapshot. Only bootstrap and sip.config may fetch the full SIP resource.
func (c *Client) ReadTask(ctx context.Context, taskID string, tenantID int64, sip SIP) (Snapshot, error) {
// ReadProviders reads the complete provider catalog once per Dispatcher boot.
// Running tasks use this immutable catalog until the next restart.
func (c *Client) ReadProviders(ctx context.Context) (map[string]Provider, error) {
var response providerResponse
if err := c.readResource(ctx, configReadPath+"/ai-providers", "ai_providers", &response); err != nil {
return nil, fmt.Errorf("read AI providers: %w", err)
}
if response.DispatcherID != c.dispatcherID {
return nil, errors.New("AI providers dispatcher owner mismatch")
}
providers := make(map[string]Provider, len(response.Providers))
for _, provider := range response.Providers {
if _, exists := providers[provider.ProviderRef]; exists {
return nil, fmt.Errorf("duplicate AI provider reference %q", provider.ProviderRef)
}
providers[provider.ProviderRef] = provider
}
return providers, nil
}
// ReadTask reads current task and quota resources using the SIP and provider
// snapshots approved at boot. No task operation fetches the global resources.
func (c *Client) ReadTask(ctx context.Context, taskID string, tenantID int64, sip SIP, providers map[string]Provider) (Snapshot, error) {
if taskID == "" || tenantID <= 0 {
return Snapshot{}, errors.New("task ID and positive tenant ID are required")
}
if sip.DispatcherID != c.dispatcherID || sip.Revision <= 0 {
return Snapshot{}, errors.New("task requires the current approved SIP snapshot")
}
result := Snapshot{SIP: sip}
var providers providerResponse
if err := c.readResource(ctx, configReadPath+"/ai-providers", "ai_providers", &providers); err != nil {
return Snapshot{}, fmt.Errorf("read AI providers: %w", err)
if sip.DispatcherID != c.dispatcherID || sip.Revision <= 0 || len(providers) == 0 {
return Snapshot{}, errors.New("task requires approved SIP and AI provider snapshots")
}
result := Snapshot{SIP: sip, Providers: providers}
if err := c.readResource(ctx, configReadPath+"/task/"+url.PathEscape(taskID), "task_config", &result.Task); err != nil {
return Snapshot{}, fmt.Errorf("read task configuration: %w", err)
}
if err := c.readResource(ctx, configReadPath+"/tenant/"+strconv.FormatInt(tenantID, 10)+"/quota", "tenant_quota", &result.Quota); err != nil {
return Snapshot{}, fmt.Errorf("read tenant quota: %w", err)
}
if result.SIP.DispatcherID != c.dispatcherID || providers.DispatcherID != c.dispatcherID || result.Task.DispatcherID != c.dispatcherID || result.Quota.DispatcherID != c.dispatcherID || result.Task.TenantID != tenantID || result.Quota.TenantID != tenantID || result.Task.TaskID != taskID {
if result.SIP.DispatcherID != c.dispatcherID || result.Task.DispatcherID != c.dispatcherID || result.Quota.DispatcherID != c.dispatcherID || result.Task.TenantID != tenantID || result.Quota.TenantID != tenantID || result.Task.TaskID != taskID {
return Snapshot{}, errors.New("configuration dispatcher, task, or tenant owner mismatch")
}
result.Providers = make(map[string]Provider, len(providers.Providers))
for _, provider := range providers.Providers {
if _, exists := result.Providers[provider.ProviderRef]; exists {
return Snapshot{}, fmt.Errorf("duplicate AI provider reference %q", provider.ProviderRef)
}
result.Providers[provider.ProviderRef] = provider
}
for _, expected := range []struct{ ref, role string }{
{result.Task.Agent.ASR.ProviderRef, "asr"},
{result.Task.Agent.LLM.ProviderRef, "llm"},
internal/configread/snapshots_test.go
+58 -5
View File
@@ -62,7 +62,11 @@ func TestReadTaskSnapshot(t *testing.T) {
if err != nil {
t.Fatal(err)
}
snapshot, err := client.ReadTask(context.Background(), "task-"+mode, 1001, approvedSIP(t))
providers, err := client.ReadProviders(context.Background())
if err != nil {
t.Fatal(err)
}
snapshot, err := client.ReadTask(context.Background(), "task-"+mode, 1001, approvedSIP(t), providers)
if err != nil {
t.Fatal(err)
}
@@ -92,6 +96,48 @@ func TestReadTaskSnapshot(t *testing.T) {
}
}
func TestReadTaskReusesBootProviders(t *testing.T) {
id := "c046b893-8628-4589-ae50-619d049248a6"
providerReads := 0
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
switch r.URL.Path {
case "/internal/v1/dispatcher/ai-providers":
providerReads++
if providerReads != 1 {
t.Errorf("AI providers fetched again after bootstrap")
w.WriteHeader(http.StatusServiceUnavailable)
return
}
_, _ = w.Write(example(t, "config-read-providers"))
case "/internal/v1/dispatcher/task/task-asr":
_, _ = w.Write(example(t, "config-read-task-asr"))
case "/internal/v1/dispatcher/tenant/1001/quota":
_, _ = w.Write(example(t, "config-read-quota"))
default:
t.Errorf("unexpected path %s", r.URL.Path)
w.WriteHeader(http.StatusNotFound)
}
}))
defer server.Close()
client, err := NewClient(server.URL, id, "test-secret", server.Client())
if err != nil {
t.Fatal(err)
}
providers, err := client.ReadProviders(context.Background())
if err != nil {
t.Fatal(err)
}
for range 2 {
if _, err := client.ReadTask(context.Background(), "task-asr", 1001, approvedSIP(t), providers); err != nil {
t.Fatal(err)
}
}
if providerReads != 1 {
t.Fatalf("provider reads=%d, want 1", providerReads)
}
}
func TestReadSIPWithoutTasks(t *testing.T) {
id := "c046b893-8628-4589-ae50-619d049248a6"
requests := 0
@@ -123,7 +169,6 @@ func TestReadTaskDoesNotReuseConfigAfterHTTPFailure(t *testing.T) {
"/internal/v1/dispatcher/tenant/1001/quota": example(t, "config-read-quota"),
}
for _, tc := range []struct{ path, owner string }{
{"/internal/v1/dispatcher/ai-providers", "AI providers"},
{"/internal/v1/dispatcher/task/task-asr", "task configuration"},
{"/internal/v1/dispatcher/tenant/1001/quota", "tenant quota"},
} {
@@ -148,11 +193,15 @@ func TestReadTaskDoesNotReuseConfigAfterHTTPFailure(t *testing.T) {
if err != nil {
t.Fatal(err)
}
if _, err := client.ReadTask(context.Background(), "task-asr", 1001, approvedSIP(t)); err != nil {
providers, err := client.ReadProviders(context.Background())
if err != nil {
t.Fatal(err)
}
if _, err := client.ReadTask(context.Background(), "task-asr", 1001, approvedSIP(t), providers); err != nil {
t.Fatalf("approved resources failed before the outage: %v", err)
}
unavailable.Store(true)
_, err = client.ReadTask(context.Background(), "task-asr", 1001, approvedSIP(t))
_, err = client.ReadTask(context.Background(), "task-asr", 1001, approvedSIP(t), providers)
var httpErr *HTTPError
if !errors.As(err, &httpErr) || httpErr.StatusCode != http.StatusServiceUnavailable {
t.Fatalf("unavailable %s reused cached configuration or lost status: %v", tc.owner, err)
@@ -192,7 +241,11 @@ func TestReadTaskRejectsMismatchedOwnerAndMalformedSIP(t *testing.T) {
if err != nil {
t.Fatal(err)
}
if _, err := client.ReadTask(context.Background(), "task-asr", 1001, approvedSIP(t)); err == nil || strings.Contains(err.Error(), "example-only-not-a-real-secret") {
providers, err := client.ReadProviders(context.Background())
if err == nil {
_, err = client.ReadTask(context.Background(), "task-asr", 1001, approvedSIP(t), providers)
}
if err == nil || strings.Contains(err.Error(), "example-only-not-a-real-secret") {
t.Fatalf("expected redacted fail-closed error, got %v", err)
}
})
internal/dispatcher/config.go
+14 -3
View File
@@ -18,8 +18,9 @@ type Bootstrap struct {
DispatcherID string
VerifySIP func(context.Context, configread.SIP) error
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
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 (b Bootstrap) Run(ctx context.Context) error {
@@ -32,6 +33,9 @@ func (b Bootstrap) Run(ctx context.Context) error {
if b.SIP != nil {
*b.SIP = configread.SIP{}
}
if b.Providers != nil {
*b.Providers = nil
}
if err := b.Store.CloseAdmission(b.DispatcherID); err != nil {
return fmt.Errorf("close task admission before bootstrap: %w", err)
}
@@ -45,6 +49,10 @@ 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)
}
providers, err := b.Client.ReadProviders(ctx)
if err != nil {
return fmt.Errorf("read boot-approved AI providers: %w", err)
}
tasks, cursor, err := b.Client.ReadAllTasks(ctx)
if err != nil {
return fmt.Errorf("retrieve complete assigned task list: %w", err)
@@ -56,7 +64,7 @@ func (b Bootstrap) Run(ctx context.Context) error {
if task.Status == "stopped" {
continue
}
snapshot, err := b.Client.ReadTask(ctx, task.TaskID, task.TenantID, sip)
snapshot, err := b.Client.ReadTask(ctx, task.TaskID, task.TenantID, sip, providers)
if err != nil {
return fmt.Errorf("read assigned task %q: %w", task.TaskID, err)
}
@@ -73,6 +81,9 @@ func (b Bootstrap) Run(ctx context.Context) error {
if b.SIP != nil {
*b.SIP = sip
}
if b.Providers != nil {
*b.Providers = providers
}
if err := b.DrainControls(ctx); err != nil {
return fmt.Errorf("drain assigned Dispatcher control queue: %w", err)
}
internal/dispatcher/control.go
+11 -10
View File
@@ -28,12 +28,13 @@ type ControlAgent interface {
}
type ControlController struct {
DispatcherID string
Store *store.Store
Client *configread.Client
Agent ControlAgent
ApprovedSIP *configread.SIP
Now func() time.Time
DispatcherID string
Store *store.Store
Client *configread.Client
Agent ControlAgent
ApprovedSIP *configread.SIP
ApprovedProviders *map[string]configread.Provider
Now func() time.Time
}
// ProcessControl applies each delivered control independently: it has no
@@ -41,8 +42,8 @@ type ControlController struct {
// barrier is durable before Agent dispatch; applied state and MQ outbox are
// committed atomically after the Agent accepts the instruction.
func (c *ControlController) ProcessControl(ctx context.Context, body []byte) error {
if c == nil || c.DispatcherID == "" || c.Store == nil || c.Client == nil || c.Agent == nil || c.ApprovedSIP == nil || c.Now == nil {
return errors.New("control processing requires Dispatcher, durable store, HTTP client, Agent, approved SIP snapshot, and clock")
if c == nil || c.DispatcherID == "" || c.Store == nil || c.Client == nil || c.Agent == nil || c.ApprovedSIP == nil || c.ApprovedProviders == nil || c.Now == nil {
return errors.New("control processing requires Dispatcher, durable store, HTTP client, Agent, approved SIP/provider snapshots, and clock")
}
if err := contract.ValidateCurrent("mq", body); err != nil {
return fmt.Errorf("invalid incoming task.control: %w", err)
@@ -80,9 +81,9 @@ func (c *ControlController) ProcessControl(ctx context.Context, body []byte) err
if policy != "" {
return errors.New("start/resume control must not carry an active-call policy")
}
// Task reads use the already approved SIP snapshot; the durable
// Task reads use the boot-approved global snapshots; the durable
// SIP and pause/stop barriers remain authoritative for admission.
snapshot, err := c.Client.ReadTask(ctx, event.Payload.TaskID, event.TenantID, *c.ApprovedSIP)
snapshot, err := c.Client.ReadTask(ctx, event.Payload.TaskID, event.TenantID, *c.ApprovedSIP, *c.ApprovedProviders)
if err != nil {
return fmt.Errorf("fresh %s task configuration: %w", event.Payload.Action, err)
}
internal/dispatcher/control_test.go
+5 -1
View File
@@ -63,6 +63,10 @@ func newControlFixture(t *testing.T) (*ControlController, *fakeControlAgent, *st
if err != nil {
t.Fatal(err)
}
providers, err := client.ReadProviders(context.Background())
if err != nil {
t.Fatal(err)
}
s, err := store.Open(filepath.Join(t.TempDir(), "state.db"))
if err != nil {
t.Fatal(err)
@@ -78,7 +82,7 @@ func newControlFixture(t *testing.T) (*ControlController, *fakeControlAgent, *st
t.Fatal(err)
}
agent := &fakeControlAgent{}
controller := &ControlController{DispatcherID: id, Store: s, Client: client, Agent: agent, ApprovedSIP: &snapshot.SIP, Now: func() time.Time { return monday(9, 30) }}
controller := &ControlController{DispatcherID: id, Store: s, Client: client, Agent: agent, ApprovedSIP: &snapshot.SIP, ApprovedProviders: &providers, Now: func() time.Time { return monday(9, 30) }}
return controller, agent, s
}
internal/dispatcher/runtime.go
+3
View File
@@ -68,8 +68,11 @@ func (r *Runtime) Serve(ctx context.Context) (result error) {
}()
var sip configread.SIP
var providers map[string]configread.Provider
r.Bootstrap.SIP = &sip
r.Bootstrap.Providers = &providers
r.Control.ApprovedSIP = &sip
r.Control.ApprovedProviders = &providers
r.Bootstrap.DrainControls = func(ctx context.Context) error {
_, err := r.Broker.DrainControlPredeclared(ctx, r.Broker.ControlQueue(), r.handleControl)
if err != nil {
internal/dispatcher/runtime_integration_test.go
+5 -1
View File
@@ -100,7 +100,7 @@ func TestRuntimeIsolatedControlBacklogExecuteAndSharedResult(t *testing.T) {
if err != nil {
t.Fatal(err)
}
var taskListReads, sipReads atomic.Int32
var taskListReads, sipReads, providerReads atomic.Int32
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
var body []byte
@@ -123,6 +123,7 @@ func TestRuntimeIsolatedControlBacklogExecuteAndSharedResult(t *testing.T) {
case "/internal/v1/dispatcher/task/task-new":
body = []byte(strings.ReplaceAll(string(configExample(t, "config-read-task-asr")), "task-asr", "task-new"))
case "/internal/v1/dispatcher/ai-providers":
providerReads.Add(1)
body = configExample(t, "config-read-providers")
case "/internal/v1/dispatcher/tenant/1001/quota":
body = configExample(t, "config-read-quota")
@@ -289,6 +290,9 @@ func TestRuntimeIsolatedControlBacklogExecuteAndSharedResult(t *testing.T) {
if got := sipReads.Load(); got != 1 {
t.Fatalf("task start/resume or idle polling reread full SIP: %d", got)
}
if got := providerReads.Load(); got != 1 {
t.Fatalf("task start/resume reread boot-approved AI providers: %d", got)
}
// A temporary task rule wait is retained durably, then its consumer
// stops so further instructions remain in SaaS's task queue. Admission
// resumes from the original identity when the configured window opens.