From a9c8afcab548a368fb521af127a74da9c2c01f53 Mon Sep 17 00:00:00 2001 From: Rogee Date: Thu, 1 Oct 2026 16:22:25 +0800 Subject: [PATCH] Load AI providers once for each Dispatcher run --- internal/configread/snapshots.go | 45 +++++++------ internal/configread/snapshots_test.go | 63 +++++++++++++++++-- internal/dispatcher/config.go | 17 ++++- internal/dispatcher/control.go | 21 ++++--- internal/dispatcher/control_test.go | 6 +- internal/dispatcher/runtime.go | 3 + .../dispatcher/runtime_integration_test.go | 6 +- 7 files changed, 123 insertions(+), 38 deletions(-) diff --git a/internal/configread/snapshots.go b/internal/configread/snapshots.go index 60b99ca..b52a3de 100644 --- a/internal/configread/snapshots.go +++ b/internal/configread/snapshots.go @@ -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"}, diff --git a/internal/configread/snapshots_test.go b/internal/configread/snapshots_test.go index f1a0e51..7597b31 100644 --- a/internal/configread/snapshots_test.go +++ b/internal/configread/snapshots_test.go @@ -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) } }) diff --git a/internal/dispatcher/config.go b/internal/dispatcher/config.go index ddc72b3..0f8ec2b 100644 --- a/internal/dispatcher/config.go +++ b/internal/dispatcher/config.go @@ -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) } diff --git a/internal/dispatcher/control.go b/internal/dispatcher/control.go index 73e71da..09bef10 100644 --- a/internal/dispatcher/control.go +++ b/internal/dispatcher/control.go @@ -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) } diff --git a/internal/dispatcher/control_test.go b/internal/dispatcher/control_test.go index 12c57fe..81c635d 100644 --- a/internal/dispatcher/control_test.go +++ b/internal/dispatcher/control_test.go @@ -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 } diff --git a/internal/dispatcher/runtime.go b/internal/dispatcher/runtime.go index b5ae468..d7cc762 100644 --- a/internal/dispatcher/runtime.go +++ b/internal/dispatcher/runtime.go @@ -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 { diff --git a/internal/dispatcher/runtime_integration_test.go b/internal/dispatcher/runtime_integration_test.go index 6646c24..63affc4 100644 --- a/internal/dispatcher/runtime_integration_test.go +++ b/internal/dispatcher/runtime_integration_test.go @@ -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.