diff --git a/internal/configread/client.go b/internal/configread/client.go index 34b9283..8459124 100644 --- a/internal/configread/client.go +++ b/internal/configread/client.go @@ -9,6 +9,7 @@ import ( "mime" "net/http" "net/url" + "strconv" "strings" "time" @@ -20,10 +21,9 @@ const dispatcherSecretHeader = "X-DISPATCHER-SECRET-KEY" const configReadPath = "/internal/v1/dispatcher" const ( - resourceSIPConfig = "sip_config" - resourceTaskConfig = "task_config" - resourceTenantQuota = "tenant_quota" - taskDiscoverySnapshot = "snapshot" + resourceSIPConfig = "sip_config" + resourceTaskConfig = "task_config" + resourceTenantQuota = "tenant_quota" ) type Client struct { @@ -37,7 +37,6 @@ type Snapshot struct { TaskID string TenantID string TenantKey string - Discovery TaskDiscovery SIPRevision int64 SIPSnapshotSHA256 string SIPCellID string @@ -62,20 +61,15 @@ type Snapshot struct { ExpiresAt time.Time SIP json.RawMessage Task json.RawMessage - Tasks json.RawMessage TenantQuota json.RawMessage } type TaskDiscovery struct { DispatcherID string - Mode string - Cursor string FromCursor string NextCursor string Tasks []DiscoveredTask - Changes []TaskDiscoveryChange Body json.RawMessage - Recovered bool } type DiscoveredTask struct { @@ -86,16 +80,6 @@ type DiscoveredTask struct { TaskRevision int64 `json:"task_revision"` } -type TaskDiscoveryChange struct { - Cursor string `json:"cursor"` - Operation string `json:"operation"` - TaskID string `json:"task_id"` - TenantID string `json:"tenant_id"` - TenantKey string `json:"tenant_key"` - Status string `json:"status"` - TaskRevision int64 `json:"task_revision"` -} - type HTTPError struct { StatusCode int Code string @@ -144,27 +128,6 @@ func (c *Client) ReadTask(ctx context.Context, taskID, tenantID string) (Snapsho return Snapshot{}, errors.New("SIP configuration identity or revision does not match the request") } - discovery, err := c.ReadTaskDiscovery(ctx, "") - if err != nil { - return Snapshot{}, err - } - if discovery.DispatcherID != c.dispatcherID || discovery.Mode != taskDiscoverySnapshot || discovery.Cursor == "" { - return Snapshot{}, errors.New("task discovery snapshot identity or mode does not match the request") - } - var discovered *DiscoveredTask - for i := range discovery.Tasks { - if discovery.Tasks[i].TaskID == taskID { - if discovered != nil { - return Snapshot{}, errors.New("task discovery contains duplicate task IDs") - } - discovered = &discovery.Tasks[i] - } - } - if discovered == nil || discovered.TenantID != tenantID { - return Snapshot{}, errors.New("task discovery does not bind the requested task to the requested tenant") - } - tasksBody := append(json.RawMessage(nil), discovery.Body...) - taskPath := configReadPath + "/task/" + url.PathEscape(taskID) taskBody, err := c.getConfig(ctx, taskPath) if err != nil { @@ -177,10 +140,7 @@ func (c *Client) ReadTask(ctx context.Context, taskID, tenantID string) (Snapsho if task.Resource != resourceTaskConfig || task.DispatcherID != c.dispatcherID || task.TaskID != taskID || task.TenantID != tenantID || task.TenantKey == "" || task.TaskRevision <= 0 || task.MaxConcurrentCalls < 0 || task.RingTimeoutMS <= 0 || task.MaxCallDurationMS <= 0 || task.Agent.AgentVersionID == "" || task.Agent.AgentVersionID != task.Agent.Config.AgentVersionID { - return Snapshot{}, errors.New("task configuration identity or limits do not match discovery") - } - if discovered.TenantKey != task.TenantKey || discovered.TaskRevision != task.TaskRevision || discovered.Status != task.Status { - return Snapshot{}, errors.New("task discovery and task configuration snapshots disagree") + return Snapshot{}, errors.New("task configuration identity or limits do not match the request") } authorizationExpiresAt, err := time.Parse(time.RFC3339, task.Agent.AuthorizationExpiresAt) if err != nil { @@ -205,7 +165,7 @@ func (c *Client) ReadTask(ctx context.Context, taskID, tenantID string) (Snapsho } return Snapshot{ - TaskID: taskID, TenantID: tenantID, TenantKey: task.TenantKey, Discovery: discovery, + TaskID: taskID, TenantID: tenantID, TenantKey: task.TenantKey, SIPRevision: sip.Revision, SIPSnapshotSHA256: sip.SnapshotSHA256, SIPCellID: sip.Artifact.CellID, SIPArtifactRevision: sip.Artifact.Revision, SIPArtifactConfigSHA256: sip.Artifact.ConfigSHA256, @@ -216,8 +176,7 @@ func (c *Client) ReadTask(ctx context.Context, taskID, tenantID string) (Snapsho AgentVersionID: task.Agent.AgentVersionID, AgentAuthorizationID: task.Agent.AuthorizationID, AgentAuthorizationExpiresAt: authorizationExpiresAt, QuotaRevision: quota.QuotaRevision, TenantMaxConcurrentCalls: int64(quota.MaxConcurrentCalls), QuotaValidUntil: quotaValidUntil, - DiscoveryCursor: discovery.Cursor, - SIP: sipBody, Task: taskBody, Tasks: tasksBody, TenantQuota: quotaBody, + SIP: sipBody, Task: taskBody, TenantQuota: quotaBody, }, nil } @@ -262,27 +221,16 @@ func (c *Client) ReadTaskStatus(ctx context.Context, taskID, tenantID, tenantKey }, nil } -// ReadTaskDiscovery accepts exactly one complete snapshot or change response. -// On HTTP 410 cursor_expired it fetches one full snapshot instead of advancing. +// ReadTaskDiscovery reads one page of latest task states after a durable event ID. +// SaaS guarantees page completeness and ordering; no per-item event ID is provided. func (c *Client) ReadTaskDiscovery(ctx context.Context, after string) (TaskDiscovery, error) { - return c.readTaskDiscovery(ctx, after, false) -} - -func (c *Client) readTaskDiscovery(ctx context.Context, after string, recovered bool) (TaskDiscovery, error) { - path := configReadPath + "/tasks" - if after != "" { - path += "?" + url.Values{"after": {after}}.Encode() + from, err := parseEventCursor(after) + if err != nil { + return TaskDiscovery{}, fmt.Errorf("invalid task-discovery after cursor: %w", err) } + path := configReadPath + "/tasks?" + url.Values{"after": {after}}.Encode() body, err := c.getTaskDiscovery(ctx, path) if err != nil { - var statusErr *HTTPError - if !recovered && after != "" && errors.As(err, &statusErr) && statusErr.StatusCode == http.StatusGone && statusErr.Code == "cursor_expired" { - full, recoveryErr := c.readTaskDiscovery(ctx, "", true) - if recoveryErr != nil { - return TaskDiscovery{}, fmt.Errorf("recover expired task-discovery cursor with full snapshot: %w", recoveryErr) - } - return full, nil - } return TaskDiscovery{}, err } var response taskDiscoveryResponse @@ -292,32 +240,36 @@ func (c *Client) readTaskDiscovery(ctx context.Context, after string, recovered if response.DispatcherID != c.dispatcherID { return TaskDiscovery{}, errors.New("task-discovery dispatcher identity does not match the request") } - result := TaskDiscovery{DispatcherID: response.DispatcherID, FromCursor: after, Body: append(json.RawMessage(nil), body...), Recovered: recovered} + to, err := parseEventCursor(response.NextCursor) + if err != nil { + return TaskDiscovery{}, fmt.Errorf("invalid task-discovery next cursor: %w", err) + } + if (len(response.Tasks) == 0 && to != from) || (len(response.Tasks) != 0 && to <= from) { + return TaskDiscovery{}, errors.New("task-discovery page cursor does not match returned tasks") + } + seen := make(map[string]struct{}, len(response.Tasks)) tenantByID, idByTenant := make(map[string]string), make(map[string]string) - if after == "" { - result.Mode, result.Cursor, result.Tasks = taskDiscoverySnapshot, response.Cursor, response.Tasks - seen := make(map[string]struct{}, len(result.Tasks)) - for _, task := range result.Tasks { - if _, duplicate := seen[task.TaskID]; duplicate { - return TaskDiscovery{}, fmt.Errorf("task discovery contains duplicate task ID %q", task.TaskID) - } - seen[task.TaskID] = struct{}{} - if err := validateTenantBinding(tenantByID, idByTenant, task.TenantID, task.TenantKey); err != nil { - return TaskDiscovery{}, err - } + for _, task := range response.Tasks { + if _, duplicate := seen[task.TaskID]; duplicate { + return TaskDiscovery{}, fmt.Errorf("task discovery contains duplicate task ID %q", task.TaskID) } - } else { - result.Mode, result.NextCursor, result.Changes = "changes", response.NextCursor, response.Changes - if (len(result.Changes) == 0 && result.NextCursor != after) || (len(result.Changes) != 0 && result.NextCursor == after) { - return TaskDiscovery{}, errors.New("task discovery change cursor does not match the complete response") - } - for _, change := range result.Changes { - if err := validateTenantBinding(tenantByID, idByTenant, change.TenantID, change.TenantKey); err != nil { - return TaskDiscovery{}, err - } + seen[task.TaskID] = struct{}{} + if err := validateTenantBinding(tenantByID, idByTenant, task.TenantID, task.TenantKey); err != nil { + return TaskDiscovery{}, err } } - return result, nil + return TaskDiscovery{ + DispatcherID: response.DispatcherID, FromCursor: after, NextCursor: response.NextCursor, + Tasks: response.Tasks, Body: append(json.RawMessage(nil), body...), + }, nil +} + +func parseEventCursor(cursor string) (uint64, error) { + value, err := strconv.ParseUint(cursor, 10, 64) + if err != nil || strconv.FormatUint(value, 10) != cursor { + return 0, errors.New("event cursor must be a canonical decimal uint64") + } + return value, nil } func validateTenantBinding(tenantByID, idByTenant map[string]string, tenantID, tenantKey string) error { @@ -422,11 +374,9 @@ type sipConfigResponse struct { } type taskDiscoveryResponse struct { - DispatcherID string `json:"dispatcher_id"` - Cursor string `json:"cursor"` - NextCursor string `json:"next_cursor"` - Tasks []DiscoveredTask `json:"tasks"` - Changes []TaskDiscoveryChange `json:"changes"` + DispatcherID string `json:"dispatcher_id"` + NextCursor string `json:"next_cursor"` + Tasks []DiscoveredTask `json:"tasks"` } type taskConfigResponse struct { diff --git a/internal/configread/client_test.go b/internal/configread/client_test.go index 2b713ba..971aafb 100644 --- a/internal/configread/client_test.go +++ b/internal/configread/client_test.go @@ -22,7 +22,7 @@ const ( mockTestSecret = "test-secret" ) -func TestClientReadTaskRequestsAndValidatesFourEndpoints(t *testing.T) { +func TestClientReadTaskRequestsAndValidatesThreeConfigEndpoints(t *testing.T) { fixtures := validConfigFixtures(t) var got []string mux := http.NewServeMux() @@ -51,7 +51,7 @@ func TestClientReadTaskRequestsAndValidatesFourEndpoints(t *testing.T) { if err != nil { t.Fatal(err) } - if len(snapshot.SIP) == 0 || len(snapshot.Task) == 0 || len(snapshot.Tasks) == 0 || len(snapshot.TenantQuota) == 0 { + if len(snapshot.SIP) == 0 || len(snapshot.Task) == 0 || len(snapshot.TenantQuota) == 0 { t.Fatalf("incomplete config snapshot: %+v", snapshot) } if snapshot.TaskID != mockTaskID || snapshot.TenantID != mockTenantID || snapshot.TenantKey != mockTenantKey { @@ -59,7 +59,6 @@ func TestClientReadTaskRequestsAndValidatesFourEndpoints(t *testing.T) { } want := []string{ "GET /internal/v1/dispatcher/sip", - "GET /internal/v1/dispatcher/tasks", "GET /internal/v1/dispatcher/task/" + mockTaskID, "GET /internal/v1/dispatcher/tenant/" + mockTenantID + "/quota", } diff --git a/internal/configread/discovery_v02_test.go b/internal/configread/discovery_v02_test.go deleted file mode 100644 index 47e0a45..0000000 --- a/internal/configread/discovery_v02_test.go +++ /dev/null @@ -1,129 +0,0 @@ -package configread - -import ( - "context" - "encoding/json" - "fmt" - "net/http" - "net/http/httptest" - "reflect" - "strings" - "testing" -) - -func TestTaskDiscoveryV02SnapshotSingleResponseAndNoQueue(t *testing.T) { - body := readConfigFixture(t, "task-discovery-snapshot-v0.2.json") - var requests []string - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - requests = append(requests, r.URL.RequestURI()) - w.Header().Set("Content-Type", "application/json") - _, _ = w.Write(body) - })) - defer server.Close() - result, err := newMockClient(t, server).ReadTaskDiscovery(context.Background(), "") - if err != nil { - t.Fatal(err) - } - if result.Mode != "snapshot" || result.Cursor != "opaque-watermark-001" || len(result.Tasks) != 1 || string(result.Body) != string(body) { - t.Fatalf("snapshot = %+v", result) - } - if !reflect.DeepEqual(requests, []string{"/internal/v1/dispatcher/tasks"}) { - t.Fatalf("requests = %v, want one request", requests) - } -} - -func TestTaskDiscoveryV02ChangesAndEmptyCursor(t *testing.T) { - for _, tc := range []struct { - name, file, after, next string - count int - }{ - {"changes", "task-discovery-changes-v0.2.json", "opaque-watermark-001", "opaque-watermark-004", 3}, - {"empty", "task-discovery-no-change-v0.2.json", "opaque-watermark-004", "opaque-watermark-004", 0}, - } { - t.Run(tc.name, func(t *testing.T) { - body := readConfigFixture(t, tc.file) - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if r.URL.Query().Get("after") != tc.after || r.URL.Query().Has("page_token") { - t.Errorf("unexpected request %s", r.URL.String()) - } - w.Header().Set("Content-Type", "application/json") - _, _ = w.Write(body) - })) - defer server.Close() - result, err := newMockClient(t, server).ReadTaskDiscovery(context.Background(), tc.after) - if err != nil { - t.Fatal(err) - } - if result.Mode != "changes" || result.NextCursor != tc.next || len(result.Changes) != tc.count || result.FromCursor != tc.after { - t.Fatalf("changes = %+v", result) - } - if tc.count != 0 && (result.Changes[2].Operation != "removed" || result.Changes[2].TaskID != "task-old") { - t.Fatalf("removal = %+v", result.Changes[2]) - } - }) - } -} - -func TestTaskDiscoveryV02ExpiredCursorFetchesOneFullSnapshot(t *testing.T) { - var requests []string - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - requests = append(requests, r.URL.RequestURI()) - w.Header().Set("Content-Type", "application/json") - if r.URL.Query().Has("after") { - w.WriteHeader(http.StatusGone) - _, _ = fmt.Fprint(w, `{"schema_version":"task-discovery.v0.2-proposal","resource":"error","error":{"code":"cursor_expired","message":"full snapshot required"}}`) - return - } - _, _ = w.Write(readConfigFixture(t, "task-discovery-snapshot-v0.2.json")) - })) - defer server.Close() - result, err := newMockClient(t, server).ReadTaskDiscovery(context.Background(), "stale") - if err != nil || !result.Recovered || result.Mode != "snapshot" || len(result.Tasks) != 1 { - t.Fatalf("recovery = %+v, %v", result, err) - } - if !reflect.DeepEqual(requests, []string{"/internal/v1/dispatcher/tasks?after=stale", "/internal/v1/dispatcher/tasks"}) { - t.Fatalf("requests = %v", requests) - } -} - -func TestTaskDiscoveryV02RejectsObsoleteFieldsAndOversize(t *testing.T) { - for _, file := range []string{"task-discovery-invalid-queue-v0.2.json", "task-discovery-invalid-page-token-v0.2.json"} { - t.Run(file, func(t *testing.T) { - body := readConfigFixture(t, file) - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { - w.Header().Set("Content-Type", "application/json") - _, _ = w.Write(body) - })) - defer server.Close() - if _, err := newMockClient(t, server).ReadTaskDiscovery(context.Background(), ""); err == nil { - t.Fatal("accepted obsolete field") - } - }) - } - for _, count := range []int{256, 257} { - t.Run(fmt.Sprint(count), func(t *testing.T) { - var base map[string]any - if err := json.Unmarshal(readConfigFixture(t, "task-discovery-snapshot-v0.2.json"), &base); err != nil { - t.Fatal(err) - } - tasks := make([]any, count) - for i := range tasks { - tasks[i] = map[string]any{"task_id": fmt.Sprintf("task-%03d", i), "tenant_id": mockTenantID, "tenant_key": mockTenantKey, "status": "running", "task_revision": 1} - } - base["tasks"] = tasks - body := marshalDiscoveryResponse(t, base) - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { - w.Header().Set("Content-Type", "application/json") - _, _ = w.Write(body) - })) - defer server.Close() - result, err := newMockClient(t, server).ReadTaskDiscovery(context.Background(), "") - if count == 256 && (err != nil || len(result.Tasks) != count) { - t.Fatalf("256 tasks: %d, %v", len(result.Tasks), err) - } - if count == 257 && (err == nil || !strings.Contains(err.Error(), "validate")) { - t.Fatalf("257 tasks: %v", err) - } - }) - } -} diff --git a/internal/configread/discovery_v03_test.go b/internal/configread/discovery_v03_test.go new file mode 100644 index 0000000..1abbaf8 --- /dev/null +++ b/internal/configread/discovery_v03_test.go @@ -0,0 +1,134 @@ +package configread + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "reflect" + "strings" + "testing" +) + +const eventTestDispatcherID = "11111111-1111-4111-8111-111111111111" + +func eventTestClient(t *testing.T, server *httptest.Server) *Client { + t.Helper() + client, err := NewClient(server.URL, eventTestDispatcherID, mockTestSecret, server.Client()) + if err != nil { + t.Fatal(err) + } + return client +} + +func TestEventCursorPagesUseOneTaskListAndAllowRepeatedTaskIDs(t *testing.T) { + pages := map[string]string{ + "0": "task-discovery-page-v0.3.json", + "2": "task-discovery-updated-v0.3.json", + "3": "task-discovery-removed-v0.3.json", + "4": "task-discovery-empty-v0.3.json", + } + var got []string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + got = append(got, r.URL.RequestURI()) + if r.Header.Get(dispatcherIDHeader) != eventTestDispatcherID || r.Header.Get(dispatcherSecretHeader) != mockTestSecret { + t.Error("Dispatcher credentials were not sent") + } + name := pages[r.URL.Query().Get("after")] + if name == "" || r.URL.Query().Has("page_token") { + t.Errorf("unexpected task request: %s", r.URL.RequestURI()) + w.WriteHeader(http.StatusBadRequest) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write(readConfigFixture(t, name)) + })) + defer server.Close() + client := eventTestClient(t, server) + cursor := "0" + var statuses []string + for range 4 { + page, err := client.ReadTaskDiscovery(context.Background(), cursor) + if err != nil { + t.Fatal(err) + } + for _, task := range page.Tasks { + statuses = append(statuses, task.TaskID+":"+task.Status) + } + cursor = page.NextCursor + } + if cursor != "4" || !reflect.DeepEqual(statuses, []string{"a01:running", "a02:paused", "a01:paused", "a02:removed"}) { + t.Fatalf("statuses=%v cursor=%q", statuses, cursor) + } + want := []string{ + "/internal/v1/dispatcher/tasks?after=0", + "/internal/v1/dispatcher/tasks?after=2", + "/internal/v1/dispatcher/tasks?after=3", + "/internal/v1/dispatcher/tasks?after=4", + } + if !reflect.DeepEqual(got, want) { + t.Fatalf("queries=%v want=%v", got, want) + } +} + +func TestEventCursorRejectsBrokenPagesWithoutFallback(t *testing.T) { + valid := string(readConfigFixture(t, "task-discovery-page-v0.3.json")) + var oversized map[string]any + if err := json.Unmarshal([]byte(valid), &oversized); err != nil { + t.Fatal(err) + } + rows := make([]any, 257) + for i := range rows { + rows[i] = oversized["tasks"].([]any)[0] + } + oversized["tasks"] = rows + oversizedBody, err := json.Marshal(oversized) + if err != nil { + t.Fatal(err) + } + cases := map[string]string{ + "nonprogress": strings.Replace(valid, `"next_cursor": "2"`, `"next_cursor": "0"`, 1), + "duplicate": strings.Replace(valid, `"task_id": "a02"`, `"task_id": "a01"`, 1), + "wrong dispatcher": strings.Replace(valid, eventTestDispatcherID, "44444444-4444-4444-8444-444444444444", 1), + "oversized page": string(oversizedBody), + "changes": string(readConfigFixture(t, "task-discovery-invalid-changes-v0.3.json")), + "cursor format": string(readConfigFixture(t, "task-discovery-invalid-cursor-v0.3.json")), + "old contract": string(readConfigFixture(t, "task-discovery-snapshot-v0.2.json")), + } + for name, body := range cases { + t.Run(name, func(t *testing.T) { + requests := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests++ + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(body)) + })) + defer server.Close() + client := eventTestClient(t, server) + if _, err := client.ReadTaskDiscovery(context.Background(), "0"); err == nil { + t.Fatal("invalid task page was accepted") + } + if requests != 1 { + t.Fatalf("requests=%d, want one without fallback", requests) + } + }) + } +} + +func TestEventCursorRejectsExpiredStatusWithoutReset(t *testing.T) { + requests := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests++ + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusGone) + _, _ = w.Write([]byte(`{"schema_version":"task-discovery.v0.3-proposal","resource":"error","error":{"code":"cursor_expired","message":"expired"}}`)) + })) + defer server.Close() + client := eventTestClient(t, server) + if _, err := client.ReadTaskDiscovery(context.Background(), "1"); err == nil { + t.Fatal("HTTP 410 accepted") + } + if requests != 1 { + t.Fatalf("requests=%d, want one without reset to initial cursor", requests) + } +} diff --git a/internal/dispatcher/config_read.go b/internal/dispatcher/config_read.go index 9ea373f..67ea50b 100644 --- a/internal/dispatcher/config_read.go +++ b/internal/dispatcher/config_read.go @@ -2,7 +2,6 @@ package dispatcher import ( "context" - "encoding/json" "errors" "fmt" "strconv" @@ -49,6 +48,13 @@ func (d *Dispatcher) LoadProjectConfig(ctx context.Context, client *configread.C if d.dispatcherID != "" && client.DispatcherID() != d.dispatcherID { return fail(errors.New("configuration client dispatcher identity mismatch")) } + assigned, err := d.store.LocalDiscoveredTaskForConfig(d.dispatcherID, taskID, tenantID) + if err != nil { + return fail(fmt.Errorf("authorize task configuration from durable discovery: %w", err)) + } + if !d.now().UTC().Before(assigned.DiscoveryObservedAt.Add(taskDiscoveryTTL)) { + return fail(errors.New("task discovery is stale; new configuration admission is closed")) + } snapshot, err := client.ReadTask(ctx, taskID, tenantID) if err != nil { return fail(err) @@ -56,19 +62,28 @@ func (d *Dispatcher) LoadProjectConfig(ctx context.Context, client *configread.C if err := verifier.VerifyAppliedSIPConfig(ctx, snapshot); err != nil { return fail(fmt.Errorf("verify Agent-applied SIP config: %w", err)) } + assigned, err = d.store.LocalDiscoveredTaskForConfig(d.dispatcherID, taskID, tenantID) + if err != nil { + return fail(fmt.Errorf("recheck task discovery after configuration read: %w", err)) + } + if snapshot.TenantKey != assigned.TenantKey || snapshot.TaskRevision != assigned.TaskRevision || snapshot.TaskStatus != assigned.Status { + return fail(errors.New("task configuration and durable discovery disagree")) + } + cursor, exists, err := d.store.LocalTaskDiscoveryCursor(d.dispatcherID) + if err != nil || !exists { + return fail(errors.New("durable task discovery cursor is unavailable")) + } now := d.now().UTC() quotaExpiry := minTime(now.Add(projectConfigTTL), snapshot.QuotaValidUntil) taskExpiry := minTime(now.Add(projectConfigTTL), snapshot.AgentAuthorizationExpiresAt) sipExpiry := now.Add(projectConfigTTL) - discoveryExpiry := now.Add(taskDiscoveryTTL) + discoveryExpiry := assigned.DiscoveryObservedAt.Add(taskDiscoveryTTL) + snapshot.DiscoveryCursor = cursor snapshot.FetchedAt = now snapshot.ExpiresAt = minTime(quotaExpiry, taskExpiry, sipExpiry, discoveryExpiry) if !snapshot.ExpiresAt.After(now) { return fail(errors.New("configuration response is expired; new admission is closed")) } - if err := d.store.ApplyLocalTaskDiscovery(localTaskDiscoveryFromRead(snapshot.Discovery)); err != nil { - return fail(fmt.Errorf("persist complete task-discovery snapshot: %w", err)) - } bundle := store.LocalConfigBundle{ DispatcherID: d.dispatcherID, TenantID: snapshot.TenantID, TenantKey: snapshot.TenantKey, TaskID: snapshot.TaskID, TenantQuotaScope: store.LocalTenantQuotaScope(d.dispatcherID, snapshot.TenantID), @@ -76,7 +91,6 @@ func (d *Dispatcher) LoadProjectConfig(ctx context.Context, client *configread.C TenantQuotaLimit: snapshot.TenantMaxConcurrentCalls, TaskQuotaLimit: snapshot.TaskMaxConcurrentCalls, Snapshots: []store.LocalConfigSnapshotRecord{ {Resource: "sip", Revision: snapshot.SIPRevision, Version: strconv.FormatInt(snapshot.SIPRevision, 10), FetchedAt: now, ExpiresAt: sipExpiry, Body: snapshot.SIP}, - {Resource: "tasks", Version: snapshot.DiscoveryCursor, FetchedAt: now, ExpiresAt: discoveryExpiry, Body: snapshot.Tasks}, {Resource: "task", TenantID: snapshot.TenantID, TaskID: snapshot.TaskID, Revision: snapshot.TaskRevision, Version: strconv.FormatInt(snapshot.TaskRevision, 10), FetchedAt: now, ExpiresAt: taskExpiry, Body: snapshot.Task}, {Resource: "tenant_quota", TenantID: snapshot.TenantID, Revision: snapshot.QuotaRevision, Version: strconv.FormatInt(snapshot.QuotaRevision, 10), FetchedAt: now, ExpiresAt: quotaExpiry, Body: snapshot.TenantQuota}, }, @@ -97,7 +111,6 @@ func (d *Dispatcher) LoadProjectConfig(ctx context.Context, client *configread.C current.SIPCellID = snapshot.SIPCellID current.SIPArtifactRevision = snapshot.SIPArtifactRevision current.SIPArtifactConfigSHA256 = snapshot.SIPArtifactConfigSHA256 - current.Tasks = append(current.Tasks[:0], snapshot.Tasks...) current.DiscoveryCursor = snapshot.DiscoveryCursor current.ExpiresAt = minTime(current.ExpiresAt, sharedExpiry) if key.tenantID == snapshot.TenantID { @@ -130,45 +143,13 @@ func (d *Dispatcher) ProjectConfigSnapshot(taskID, tenantID string) (configread. } func cloneConfigSnapshot(snapshot configread.Snapshot) configread.Snapshot { - snapshot.Discovery.Body = append(json.RawMessage(nil), snapshot.Discovery.Body...) - snapshot.Discovery.Tasks = append([]configread.DiscoveredTask(nil), snapshot.Discovery.Tasks...) - snapshot.Discovery.Changes = append([]configread.TaskDiscoveryChange(nil), snapshot.Discovery.Changes...) snapshot.SIP = append(snapshot.SIP[:0:0], snapshot.SIP...) snapshot.Task = append(snapshot.Task[:0:0], snapshot.Task...) - snapshot.Tasks = append(snapshot.Tasks[:0:0], snapshot.Tasks...) snapshot.TenantQuota = append(snapshot.TenantQuota[:0:0], snapshot.TenantQuota...) snapshot.TaskAllowedTrunkIDs = append([]string(nil), snapshot.TaskAllowedTrunkIDs...) return snapshot } -func localTaskDiscoveryFromRead(discovery configread.TaskDiscovery) store.LocalTaskDiscovery { - local := store.LocalTaskDiscovery{ - DispatcherID: discovery.DispatcherID, Mode: discovery.Mode, - Cursor: discovery.Cursor, FromCursor: discovery.FromCursor, - NextCursor: discovery.NextCursor, Body: append([]byte(nil), discovery.Body...), - } - for _, task := range discovery.Tasks { - local.Tasks = append(local.Tasks, store.LocalDiscoveredTask{ - TaskID: task.TaskID, TenantID: task.TenantID, TenantKey: task.TenantKey, - Status: task.Status, TaskRevision: task.TaskRevision, - }) - } - for _, change := range discovery.Changes { - localChange := store.LocalTaskDiscoveryChange{ - Cursor: change.Cursor, Operation: change.Operation, TaskID: change.TaskID, - TenantID: change.TenantID, TenantKey: change.TenantKey, - } - if change.Operation != "removed" { - localChange.Task = store.LocalDiscoveredTask{ - TaskID: change.TaskID, TenantID: change.TenantID, TenantKey: change.TenantKey, - Status: change.Status, TaskRevision: change.TaskRevision, - } - } - local.Changes = append(local.Changes, localChange) - } - return local -} - func minTime(times ...time.Time) time.Time { if len(times) == 0 { return time.Time{} diff --git a/internal/dispatcher/local_config_read_test.go b/internal/dispatcher/local_config_read_test.go index e37cd3f..d75cf28 100644 --- a/internal/dispatcher/local_config_read_test.go +++ b/internal/dispatcher/local_config_read_test.go @@ -33,7 +33,7 @@ func TestDispatcherLoadsAndPersistsLocalConfigReadSnapshot(t *testing.T) { t.Fatal("dispatcher did not retain an unexpired config snapshot") } if snapshot.TaskID != localTestTaskID || snapshot.TenantID != localTestTenantID || - len(snapshot.SIP) == 0 || len(snapshot.Tasks) == 0 || len(snapshot.Task) == 0 || len(snapshot.TenantQuota) == 0 { + len(snapshot.SIP) == 0 || len(snapshot.Task) == 0 || len(snapshot.TenantQuota) == 0 { t.Fatalf("incomplete configuration snapshot: %+v", snapshot) } persisted, err := st.LocalConfigSnapshot(localTestDispatcherID, "task", localTestTenantID, localTestTaskID) @@ -139,6 +139,7 @@ func TestDispatcherSharesUpdatedTenantQuotaAcrossTasks(t *testing.T) { if err != nil { t.Fatal(err) } + seedLocalEventDiscovery(t, st, now, "task-b") if err := d.LoadProjectConfig(context.Background(), client, localTestSIPConfigVerifier(), "task-b", localTestTenantID); err != nil { t.Fatal(err) } diff --git a/internal/dispatcher/local_v01.go b/internal/dispatcher/local_v01.go index b4872e0..4a6c17b 100644 --- a/internal/dispatcher/local_v01.go +++ b/internal/dispatcher/local_v01.go @@ -192,10 +192,10 @@ func (d *Dispatcher) AcceptLocalV01Command(ctx context.Context, routingKey strin store.LocalTaskQuotaScope(d.dispatcherID, command.TenantID, projectConfig.TaskID), }, Snapshot: store.LocalExecutionConfigSnapshot{ - SchemaVersion: "execution-config-snapshot.v0.2", DispatcherID: d.dispatcherID, + SchemaVersion: "execution-config-snapshot.v0.3", DispatcherID: d.dispatcherID, TenantID: projectConfig.TenantID, TenantKey: projectConfig.TenantKey, TaskID: projectConfig.TaskID, SelectedTrunkID: dial.TrunkID, SIPRevision: projectConfig.SIPRevision, TaskRevision: projectConfig.TaskRevision, QuotaRevision: projectConfig.QuotaRevision, - SIP: projectConfig.SIP, Tasks: projectConfig.Tasks, Task: projectConfig.Task, TenantQuota: projectConfig.TenantQuota, + SIP: projectConfig.SIP, Task: projectConfig.Task, TenantQuota: projectConfig.TenantQuota, }, } capacityRejected := event diff --git a/internal/dispatcher/local_v01_integration_test.go b/internal/dispatcher/local_v01_integration_test.go index 30e815b..6a29eda 100644 --- a/internal/dispatcher/local_v01_integration_test.go +++ b/internal/dispatcher/local_v01_integration_test.go @@ -35,7 +35,6 @@ func TestLocalDispatcherMockSaaSEndToEndWithOutboxRecovery(t *testing.T) { // only this isolated Mock grants a positive, explicit trunk limit. fixtures := map[string][]byte{ "/internal/v1/dispatcher/sip": localConfigFixture(t, "config-read-sip-v0.1.json"), - "/internal/v1/dispatcher/tasks": readLocalFixture(t, "task-discovery-snapshot-v0.2.json"), "/internal/v1/dispatcher/task/" + localTestTaskID: readLocalFixture(t, "config-read-task-v0.1.json"), "/internal/v1/dispatcher/tenant/" + localTestTenantID + "/quota": readLocalFixture(t, "config-read-tenant-quota-v0.1.json"), } @@ -50,6 +49,18 @@ func TestLocalDispatcherMockSaaSEndToEndWithOutboxRecovery(t *testing.T) { _, _ = fmt.Fprint(w, `{"schema_version":"config-read.v0.1","resource":"error","error":{"code":"unauthorized","message":"unauthorized"}}`) return } + if r.URL.Path == "/internal/v1/dispatcher/tasks" { + w.Header().Set("Content-Type", "application/json") + switch r.URL.Query().Get("after") { + case "0": + fmt.Fprintf(w, `{"schema_version":"task-discovery.v0.3-proposal","dispatcher_id":%q,"tasks":[{"task_id":%q,"tenant_id":%q,"tenant_key":%q,"status":"running","task_revision":2}],"next_cursor":"1"}`, localTestDispatcherID, localTestTaskID, localTestTenantID, localTestTenantKey) + case "1": + fmt.Fprintf(w, `{"schema_version":"task-discovery.v0.3-proposal","dispatcher_id":%q,"tasks":[],"next_cursor":"1"}`, localTestDispatcherID) + default: + w.WriteHeader(http.StatusBadRequest) + } + return + } body, ok := fixtures[r.URL.Path] if !ok { w.WriteHeader(http.StatusNotFound) @@ -140,6 +151,10 @@ func TestLocalDispatcherMockSaaSEndToEndWithOutboxRecovery(t *testing.T) { if err != nil { t.Fatal(err) } + runtime := &LocalV01Runtime{dispatcher: d, client: configClient} + if err := runtime.refreshDiscoveryState(context.Background()); err != nil { + t.Fatal(err) + } if err := d.LoadProjectConfig(context.Background(), configClient, localTestSIPConfigVerifier(), localTestTaskID, localTestTenantID); err != nil { t.Fatal(err) } @@ -147,8 +162,9 @@ func TestLocalDispatcherMockSaaSEndToEndWithOutboxRecovery(t *testing.T) { gotRequests := append([]string(nil), requests...) requestsMu.Unlock() wantRequests := []string{ + "GET /internal/v1/dispatcher/tasks?after=0", + "GET /internal/v1/dispatcher/tasks?after=1", "GET /internal/v1/dispatcher/sip", - "GET /internal/v1/dispatcher/tasks", "GET /internal/v1/dispatcher/task/" + localTestTaskID, "GET /internal/v1/dispatcher/tenant/" + localTestTenantID + "/quota", } diff --git a/internal/dispatcher/local_v01_test.go b/internal/dispatcher/local_v01_test.go index a41dd71..386ca8c 100644 --- a/internal/dispatcher/local_v01_test.go +++ b/internal/dispatcher/local_v01_test.go @@ -8,6 +8,7 @@ import ( "net/http/httptest" "os" "path/filepath" + "strconv" "strings" "testing" "time" @@ -78,7 +79,7 @@ func TestAcceptLocalV01CommandPersistsReceiptAndRejectsExpiredCommand(t *testing t.Fatal(err) } var bound store.LocalExecutionConfigSnapshot - if err := json.Unmarshal(executionBody, &bound); err != nil || bound.SelectedTrunkID != "trunk-mock" || bound.SchemaVersion != "execution-config-snapshot.v0.2" { + if err := json.Unmarshal(executionBody, &bound); err != nil || bound.SelectedTrunkID != "trunk-mock" || bound.SchemaVersion != "execution-config-snapshot.v0.3" { t.Fatalf("accepted call did not bind its selected trunk: %+v err=%v", bound, err) } if err := st.MarkOutboxPublished(accepted[0].ID); err != nil { @@ -517,7 +518,16 @@ func newLocalV01TestDispatcher(t *testing.T, now time.Time) (*Dispatcher, *store t.Helper() mux := http.NewServeMux() mux.HandleFunc("/internal/v1/dispatcher/sip", localConfigResponse(string(localConfigFixture(t, "config-read-sip-v0.1.json")))) - mux.HandleFunc("/internal/v1/dispatcher/tasks", localConfigResponse(string(localConfigFixture(t, "task-discovery-snapshot-v0.2.json")))) + mux.HandleFunc("/internal/v1/dispatcher/tasks", func(w http.ResponseWriter, r *http.Request) { + after := r.URL.Query().Get("after") + if after == "" || after == "0" { + t.Errorf("expected durable event cursor, got %s", r.URL.RequestURI()) + w.WriteHeader(http.StatusBadRequest) + return + } + w.Header().Set("Content-Type", "application/json") + fmt.Fprintf(w, `{"schema_version":"task-discovery.v0.3-proposal","dispatcher_id":"%s","tasks":[],"next_cursor":%q}`, localTestDispatcherID, after) + }) mux.HandleFunc("/internal/v1/dispatcher/task/"+localTestTaskID, localConfigResponse(string(localConfigFixture(t, "config-read-task-v0.1.json")))) mux.HandleFunc("/internal/v1/dispatcher/tenant/"+localTestTenantID+"/quota", localConfigResponse(string(localConfigFixture(t, "config-read-tenant-quota-v0.1.json")))) server := httptest.NewServer(mux) @@ -538,6 +548,7 @@ func newLocalV01TestDispatcher(t *testing.T, now time.Time) (*Dispatcher, *store server.Close() t.Fatal(err) } + seedLocalEventDiscovery(t, st, now, localTestTaskID) if err := d.LoadProjectConfig(context.Background(), client, localTestSIPConfigVerifier(), localTestTaskID, localTestTenantID); err != nil { st.Close() server.Close() @@ -546,6 +557,42 @@ func newLocalV01TestDispatcher(t *testing.T, now time.Time) (*Dispatcher, *store return d, st, server } +func seedLocalEventDiscovery(t *testing.T, st *store.Store, now time.Time, taskIDs ...string) { + t.Helper() + var tasks []store.LocalDiscoveredTask + var wire []map[string]any + for _, taskID := range taskIDs { + tasks = append(tasks, store.LocalDiscoveredTask{TaskID: taskID, TenantID: localTestTenantID, TenantKey: localTestTenantKey, Status: "running", TaskRevision: 2}) + wire = append(wire, map[string]any{"task_id": taskID, "tenant_id": localTestTenantID, "tenant_key": localTestTenantKey, "status": "running", "task_revision": 2}) + } + after, exists, err := st.LocalTaskDiscoveryCursor(localTestDispatcherID) + if err != nil { + t.Fatal(err) + } + if !exists { + after = "0" + } + position, err := strconv.ParseUint(after, 10, 64) + if err != nil || position == ^uint64(0) { + t.Fatalf("invalid test discovery cursor %q: %v", after, err) + } + next := strconv.FormatUint(position+1, 10) + body, err := json.Marshal(map[string]any{"schema_version": "task-discovery.v0.3-proposal", "dispatcher_id": localTestDispatcherID, "tasks": wire, "next_cursor": next}) + if err != nil { + t.Fatal(err) + } + if err := st.ApplyLocalTaskDiscoveryPage(store.LocalTaskDiscoveryPage{DispatcherID: localTestDispatcherID, FromCursor: after, NextCursor: next, Tasks: tasks, Body: body, ObservedAt: now}); err != nil { + t.Fatal(err) + } + empty, err := json.Marshal(map[string]any{"schema_version": "task-discovery.v0.3-proposal", "dispatcher_id": localTestDispatcherID, "tasks": []any{}, "next_cursor": next}) + if err != nil { + t.Fatal(err) + } + if err := st.ApplyLocalTaskDiscoveryPage(store.LocalTaskDiscoveryPage{DispatcherID: localTestDispatcherID, FromCursor: next, NextCursor: next, Body: empty, ObservedAt: now}); err != nil { + t.Fatal(err) + } +} + func newLocalV01MultiTaskConfigServer(t *testing.T, taskIDs []string, quotaRevision, quotaLimit int64) *httptest.Server { t.Helper() if len(taskIDs) == 0 { diff --git a/internal/dispatcher/task_runtime_v03_recovery_test.go b/internal/dispatcher/task_runtime_v03_recovery_test.go new file mode 100644 index 0000000..ddb45ac --- /dev/null +++ b/internal/dispatcher/task_runtime_v03_recovery_test.go @@ -0,0 +1,143 @@ +package dispatcher + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "path/filepath" + "reflect" + "sync" + "sync/atomic" + "testing" +) + +func TestEventRuntimeResumesAtCommittedCursorAfterRestart(t *testing.T) { + var fail atomic.Bool + fail.Store(true) + var mu sync.Mutex + var requests []string + pages := map[string]string{ + "0": "task-discovery-page-v0.3.json", + "2": "task-discovery-updated-v0.3.json", + "3": "task-discovery-removed-v0.3.json", + "4": "task-discovery-empty-v0.3.json", + } + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + after := r.URL.Query().Get("after") + mu.Lock() + requests = append(requests, after) + mu.Unlock() + w.Header().Set("Content-Type", "application/json") + if r.URL.Path != "/internal/v1/dispatcher/tasks" || pages[after] == "" { + w.WriteHeader(http.StatusBadRequest) + return + } + if after == "2" && fail.Load() { + w.WriteHeader(http.StatusServiceUnavailable) + _, _ = w.Write([]byte(`{"schema_version":"task-discovery.v0.3-proposal","resource":"error","error":{"code":"service_unavailable","message":"mock outage"}}`)) + return + } + _, _ = w.Write(eventRuntimeFixture(t, pages[after])) + })) + defer server.Close() + path := filepath.Join(t.TempDir(), "dispatcher.db") + runtime, st := newEventRuntimeAt(t, server, path) + if err := runtime.refreshDiscoveryState(context.Background()); err == nil { + t.Fatal("partial discovery accepted before restart") + } + cursor, exists, err := st.LocalTaskDiscoveryCursor(localTestDispatcherID) + if err != nil || !exists || cursor != "2" { + t.Fatalf("checkpoint before restart=%q exists=%v err=%v", cursor, exists, err) + } + if err := st.Close(); err != nil { + t.Fatal(err) + } + fail.Store(false) + restarted, reopened := newEventRuntimeAt(t, server, path) + if err := restarted.refreshDiscoveryState(context.Background()); err != nil { + t.Fatal(err) + } + mu.Lock() + got := append([]string(nil), requests...) + mu.Unlock() + if !reflect.DeepEqual(got, []string{"0", "2", "2", "3", "4"}) { + t.Fatalf("restart reset or skipped event cursor: requests=%v", got) + } + var ready int + if err := reopened.DB().QueryRow(`SELECT cursor,ready FROM local_v03_task_discovery_state WHERE dispatcher_id=?`, localTestDispatcherID).Scan(&cursor, &ready); err != nil || cursor != "4" || ready != 1 { + t.Fatalf("checkpoint after restart=%q ready=%d err=%v", cursor, ready, err) + } + assignments, err := reopened.LocalTaskAssignments(localTestDispatcherID) + if err != nil || len(assignments) != 2 || assignments[0].Status != "paused" || !assignments[1].Removed { + t.Fatalf("replayed assignments=%+v err=%v", assignments, err) + } +} + +func TestEventRuntimeCatchesUpdateArrivingDuringPagination(t *testing.T) { + firstPageRequested := make(chan struct{}) + updateCommitted := make(chan struct{}) + var updated atomic.Bool + go func() { + <-firstPageRequested + updated.Store(true) + close(updateCommitted) + }() + var mu sync.Mutex + var requests []string + task := func(id, status string) string { + return fmt.Sprintf(`{"task_id":%q,"tenant_id":%q,"tenant_key":%q,"status":%q,"task_revision":2}`, id, localTestTenantID, localTestTenantKey, status) + } + page := func(tasks, next string) []byte { + return []byte(fmt.Sprintf(`{"schema_version":"task-discovery.v0.3-proposal","dispatcher_id":%q,"tasks":[%s],"next_cursor":%q}`, localTestDispatcherID, tasks, next)) + } + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + after := r.URL.Query().Get("after") + mu.Lock() + requests = append(requests, after) + mu.Unlock() + w.Header().Set("Content-Type", "application/json") + if r.URL.Path != "/internal/v1/dispatcher/tasks" { + w.WriteHeader(http.StatusBadRequest) + return + } + switch after { + case "0": + body := page(task(localTestTaskID, "running"), "1") + close(firstPageRequested) + <-updateCommitted + _, _ = w.Write(body) + case "1": + _, _ = w.Write(page(task("task-b", "running"), "2")) + case "2": + if !updated.Load() { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + _, _ = w.Write(page(task(localTestTaskID, "paused"), "3")) + case "3": + _, _ = w.Write(page("", "3")) + default: + w.WriteHeader(http.StatusBadRequest) + } + })) + defer server.Close() + runtime, st := newEventRuntime(t, server) + if err := runtime.refreshDiscoveryState(context.Background()); err != nil { + t.Fatal(err) + } + mu.Lock() + got := append([]string(nil), requests...) + mu.Unlock() + if !reflect.DeepEqual(got, []string{"0", "1", "2", "3"}) { + t.Fatalf("concurrent update lost or pages skipped: %v", got) + } + assignments, err := st.LocalTaskAssignments(localTestDispatcherID) + if err != nil || len(assignments) != 2 { + t.Fatalf("incomplete assignments: %+v err=%v", assignments, err) + } + status := map[string]string{assignments[0].TaskID: assignments[0].Status, assignments[1].TaskID: assignments[1].Status} + if status[localTestTaskID] != "paused" || status["task-b"] != "running" { + t.Fatalf("update did not supersede earlier page: %+v", assignments) + } +} diff --git a/internal/dispatcher/task_runtime_v03_test.go b/internal/dispatcher/task_runtime_v03_test.go new file mode 100644 index 0000000..2f4237c --- /dev/null +++ b/internal/dispatcher/task_runtime_v03_test.go @@ -0,0 +1,125 @@ +package dispatcher + +import ( + "bytes" + "context" + "net/http" + "net/http/httptest" + "path/filepath" + "reflect" + "sync/atomic" + "testing" + "time" + + "git.ipao.vip/rogee/go-sip/internal/configread" + "git.ipao.vip/rogee/go-sip/internal/store" +) + +func eventRuntimeFixture(t *testing.T, name string) []byte { + t.Helper() + return bytes.ReplaceAll(localConfigFixture(t, name), []byte("11111111-1111-4111-8111-111111111111"), []byte(localTestDispatcherID)) +} + +func newEventRuntime(t *testing.T, server *httptest.Server) (*LocalV01Runtime, *store.Store) { + t.Helper() + return newEventRuntimeAt(t, server, filepath.Join(t.TempDir(), "dispatcher.db")) +} + +func newEventRuntimeAt(t *testing.T, server *httptest.Server, path string) (*LocalV01Runtime, *store.Store) { + t.Helper() + st, err := store.Open(path) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = st.Close() }) + d, err := NewV3(localTestDispatcherID, st, nil, time.Now) + if err != nil { + t.Fatal(err) + } + client, err := configread.NewClient(server.URL, localTestDispatcherID, "test-secret", server.Client()) + if err != nil { + t.Fatal(err) + } + return &LocalV01Runtime{dispatcher: d, client: client}, st +} + +func TestEventRuntimeBootsOnlyAfterAllPagesAndEmptyPage(t *testing.T) { + pages := map[string]string{"0": "task-discovery-page-v0.3.json", "2": "task-discovery-updated-v0.3.json", "3": "task-discovery-removed-v0.3.json", "4": "task-discovery-empty-v0.3.json"} + var afters []string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + after := r.URL.Query().Get("after") + afters = append(afters, after) + if r.URL.Path != "/internal/v1/dispatcher/tasks" || r.URL.Query().Has("page_token") || pages[after] == "" { + t.Errorf("bad discovery request: %s", r.URL.RequestURI()) + w.WriteHeader(http.StatusBadRequest) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write(eventRuntimeFixture(t, pages[after])) + })) + defer server.Close() + runtime, st := newEventRuntime(t, server) + if err := runtime.refreshDiscoveryState(context.Background()); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(afters, []string{"0", "2", "3", "4"}) { + t.Fatalf("pages read=%v", afters) + } + var ready int + if err := st.DB().QueryRow(`SELECT ready FROM local_v03_task_discovery_state WHERE dispatcher_id=?`, localTestDispatcherID).Scan(&ready); err != nil || ready != 1 { + t.Fatalf("complete discovery ready=%d err=%v", ready, err) + } + assignments, err := st.LocalTaskAssignments(localTestDispatcherID) + if err != nil || len(assignments) != 2 || assignments[0].Status != "paused" || assignments[1].Status != "removed" { + t.Fatalf("assignments=%+v err=%v", assignments, err) + } + if err := runtime.refreshDiscoveryState(context.Background()); err != nil { + t.Fatal(err) + } + if afters[len(afters)-1] != "4" { + t.Fatalf("poll did not resume durable cursor: %v", afters) + } +} + +func TestEventRuntimeInterruptedPageResumesWithoutReset(t *testing.T) { + var fail atomic.Bool + fail.Store(true) + var afters []string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + after := r.URL.Query().Get("after") + afters = append(afters, after) + w.Header().Set("Content-Type", "application/json") + if after == "2" && fail.Load() { + w.WriteHeader(http.StatusServiceUnavailable) + _, _ = w.Write([]byte(`{"schema_version":"task-discovery.v0.3-proposal","resource":"error","error":{"code":"service_unavailable","message":"test outage"}}`)) + return + } + name := map[string]string{"0": "task-discovery-page-v0.3.json", "2": "task-discovery-updated-v0.3.json", "3": "task-discovery-removed-v0.3.json", "4": "task-discovery-empty-v0.3.json"}[after] + if name == "" { + t.Errorf("unexpected cursor %q", after) + w.WriteHeader(http.StatusBadRequest) + return + } + _, _ = w.Write(eventRuntimeFixture(t, name)) + })) + defer server.Close() + runtime, st := newEventRuntime(t, server) + if err := runtime.refreshDiscoveryState(context.Background()); err == nil { + t.Fatal("partial page accepted") + } + cursor, exists, err := st.LocalTaskDiscoveryCursor(localTestDispatcherID) + if err != nil || !exists || cursor != "2" { + t.Fatalf("partial cursor=%q exists=%v err=%v", cursor, exists, err) + } + var ready int + if err := st.DB().QueryRow(`SELECT ready FROM local_v03_task_discovery_state WHERE dispatcher_id=?`, localTestDispatcherID).Scan(&ready); err != nil || ready != 0 { + t.Fatalf("partial response opened admission: ready=%d err=%v", ready, err) + } + fail.Store(false) + if err := runtime.refreshDiscoveryState(context.Background()); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(afters, []string{"0", "2", "2", "3", "4"}) { + t.Fatalf("retry must use persisted cursor, got %v", afters) + } +} diff --git a/internal/dispatcher/task_runtime_v3.go b/internal/dispatcher/task_runtime_v3.go index 6cba4c1..5136f0d 100644 --- a/internal/dispatcher/task_runtime_v3.go +++ b/internal/dispatcher/task_runtime_v3.go @@ -153,17 +153,35 @@ func (r *LocalV01Runtime) refreshDiscoveryState(ctx context.Context) error { if err != nil { return r.failDiscovery(fmt.Errorf("read durable task-discovery cursor: %w", err)) } + if err := r.dispatcher.store.CloseLocalTaskDiscoveryAdmission(r.dispatcher.dispatcherID); err != nil { + return r.failDiscovery(fmt.Errorf("close task admission before discovery: %w", err)) + } if !exists { - cursor = "" + cursor = "0" } - discovery, err := r.client.ReadTaskDiscovery(ctx, cursor) - if err != nil { - return r.failDiscovery(fmt.Errorf("read task discovery: %w", err)) + pageCtx, cancel := context.WithTimeout(ctx, taskDiscoveryPollInterval) + defer cancel() + for { + page, err := r.client.ReadTaskDiscovery(pageCtx, cursor) + if err != nil { + return r.failDiscovery(fmt.Errorf("read task discovery after %s: %w", cursor, err)) + } + tasks := make([]store.LocalDiscoveredTask, 0, len(page.Tasks)) + for _, task := range page.Tasks { + tasks = append(tasks, store.LocalDiscoveredTask{TaskID: task.TaskID, TenantID: task.TenantID, + TenantKey: task.TenantKey, TaskRevision: task.TaskRevision, Status: task.Status}) + } + if err := r.dispatcher.store.ApplyLocalTaskDiscoveryPage(store.LocalTaskDiscoveryPage{ + DispatcherID: page.DispatcherID, FromCursor: cursor, NextCursor: page.NextCursor, + Tasks: tasks, Body: page.Body, ObservedAt: r.dispatcher.now().UTC(), + }); err != nil { + return r.failDiscovery(fmt.Errorf("persist task-discovery page after %s: %w", cursor, err)) + } + if len(tasks) == 0 { + return nil + } + cursor = page.NextCursor } - if err := r.dispatcher.store.ApplyLocalTaskDiscovery(localTaskDiscoveryFromRead(discovery)); err != nil { - return r.failDiscovery(fmt.Errorf("persist complete task-discovery response: %w", err)) - } - return nil } func (r *LocalV01Runtime) failDiscovery(cause error) error { diff --git a/internal/dispatcher/task_runtime_v3_test.go b/internal/dispatcher/task_runtime_v3_test.go index cf49ae3..885842c 100644 --- a/internal/dispatcher/task_runtime_v3_test.go +++ b/internal/dispatcher/task_runtime_v3_test.go @@ -2,6 +2,7 @@ package dispatcher import ( "context" + "fmt" "net/http" "net/http/httptest" "path/filepath" @@ -13,64 +14,33 @@ import ( "git.ipao.vip/rogee/go-sip/internal/store" ) -func TestLocalV01RuntimeRefreshDiscoveryPersistsCompleteSnapshotAndCursor(t *testing.T) { - body := localConfigFixture(t, "task-discovery-snapshot-v0.2.json") - var afterValues []string - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodGet || r.URL.Path != "/internal/v1/dispatcher/tasks" { - t.Errorf("discovery request = %s %s", r.Method, r.URL.RequestURI()) - } - afterValues = append(afterValues, r.URL.Query().Get("after")) - w.Header().Set("Content-Type", "application/json") - _, _ = w.Write(body) - })) - defer server.Close() - st, err := store.Open(filepath.Join(t.TempDir(), "dispatcher.db")) - if err != nil { - t.Fatal(err) - } - defer st.Close() - d, err := NewV3(localTestDispatcherID, st, nil, time.Now) - if err != nil { - t.Fatal(err) - } - client, err := configread.NewClient(server.URL, localTestDispatcherID, "test-secret", server.Client()) - if err != nil { - t.Fatal(err) - } - runtime := &LocalV01Runtime{dispatcher: d, client: client} - if err := runtime.refreshDiscoveryState(context.Background()); err != nil { - t.Fatal(err) - } - if len(afterValues) != 1 || afterValues[0] != "" { - t.Fatalf("initial discovery cursors = %v, want one full-snapshot request", afterValues) - } - cursor, exists, err := st.LocalTaskDiscoveryCursor(localTestDispatcherID) - if err != nil || !exists || cursor == "" { - t.Fatalf("persisted cursor=%q exists=%v err=%v", cursor, exists, err) - } - assignments, err := st.LocalTaskAssignments(localTestDispatcherID) - if err != nil || len(assignments) != 1 || assignments[0].TaskID != localTestTaskID { - t.Fatalf("persisted assignments=%+v err=%v", assignments, err) - } -} - func TestLocalV01RuntimeDiscoveryFailureClosesAdmissionAndStopsTaskConsumers(t *testing.T) { var phase atomic.Int32 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + after := r.URL.Query().Get("after") if r.URL.Path != "/internal/v1/dispatcher/tasks" { t.Errorf("unexpected path: %s", r.URL.Path) + w.WriteHeader(http.StatusNotFound) + return } - w.Header().Set("Content-Type", "application/json") switch phase.Load() { case 0: - _, _ = w.Write(localConfigFixture(t, "task-discovery-snapshot-v0.2.json")) + if after == "0" { + fmt.Fprintf(w, `{"schema_version":"task-discovery.v0.3-proposal","dispatcher_id":%q,"tasks":[{"task_id":%q,"tenant_id":%q,"tenant_key":%q,"status":"running","task_revision":2}],"next_cursor":"1"}`, localTestDispatcherID, localTestTaskID, localTestTenantID, localTestTenantKey) + return + } case 1: w.WriteHeader(http.StatusServiceUnavailable) - _, _ = w.Write([]byte(`{"schema_version":"task-discovery.v0.2-proposal","resource":"error","error":{"code":"service_unavailable","message":"mock outage"}}`)) - default: - _, _ = w.Write([]byte(`{"schema_version":"task-discovery.v0.2-proposal","dispatcher_id":"c046b893-8628-4589-ae50-619d049248a6","next_cursor":"opaque-watermark-001","changes":[]}`)) + _, _ = w.Write([]byte(`{"schema_version":"task-discovery.v0.3-proposal","resource":"error","error":{"code":"service_unavailable","message":"mock outage"}}`)) + return } + if after != "1" { + t.Errorf("unexpected event cursor: %q", after) + w.WriteHeader(http.StatusBadRequest) + return + } + fmt.Fprintf(w, `{"schema_version":"task-discovery.v0.3-proposal","dispatcher_id":%q,"tasks":[],"next_cursor":"1"}`, localTestDispatcherID) })) defer server.Close() st, err := store.Open(filepath.Join(t.TempDir(), "dispatcher.db")) @@ -104,7 +74,7 @@ func TestLocalV01RuntimeDiscoveryFailureClosesAdmissionAndStopsTaskConsumers(t * t.Fatal("outage accepted as fresh discovery") } var ready int - if err := st.DB().QueryRow(`SELECT ready FROM local_v02_task_discovery_state WHERE dispatcher_id=?`, localTestDispatcherID).Scan(&ready); err != nil || ready != 0 { + if err := st.DB().QueryRow(`SELECT ready FROM local_v03_task_discovery_state WHERE dispatcher_id=?`, localTestDispatcherID).Scan(&ready); err != nil || ready != 0 { t.Fatalf("discovery gate after outage=%d err=%v", ready, err) } select { @@ -116,84 +86,8 @@ func TestLocalV01RuntimeDiscoveryFailureClosesAdmissionAndStopsTaskConsumers(t * if err := runtime.refreshDiscoveryState(context.Background()); err != nil { t.Fatal(err) } - if err := st.DB().QueryRow(`SELECT ready FROM local_v02_task_discovery_state WHERE dispatcher_id=?`, localTestDispatcherID).Scan(&ready); err != nil || ready != 1 { - t.Fatalf("discovery gate after fresh no-change=%d err=%v", ready, err) - } -} - -func TestLocalV01RuntimeExpiredCursorRecoversFullSnapshotAndRetiresTask(t *testing.T) { - var calls atomic.Int32 - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if r.URL.Path != "/internal/v1/dispatcher/tasks" { - t.Errorf("unexpected path: %s", r.URL.Path) - } - w.Header().Set("Content-Type", "application/json") - switch calls.Add(1) { - case 1: - if r.URL.Query().Has("after") { - t.Error("initial read must be a full snapshot") - } - _, _ = w.Write(localConfigFixture(t, "task-discovery-snapshot-v0.2.json")) - case 2: - if r.URL.Query().Get("after") != "opaque-watermark-001" { - t.Errorf("incremental cursor = %s", r.URL.RawQuery) - } - w.WriteHeader(http.StatusGone) - _, _ = w.Write([]byte(`{"schema_version":"task-discovery.v0.2-proposal","resource":"error","error":{"code":"cursor_expired","message":"mock expired"}}`)) - case 3: - if r.URL.Query().Has("after") { - t.Errorf("recovery must be a full snapshot: %s", r.URL.RawQuery) - } - _, _ = w.Write([]byte(`{"schema_version":"task-discovery.v0.2-proposal","dispatcher_id":"c046b893-8628-4589-ae50-619d049248a6","cursor":"opaque-watermark-002","tasks":[]}`)) - default: - t.Error("unexpected fourth request") - } - })) - defer server.Close() - st, err := store.Open(filepath.Join(t.TempDir(), "dispatcher.db")) - if err != nil { - t.Fatal(err) - } - defer st.Close() - d, err := NewV3(localTestDispatcherID, st, nil, time.Now) - if err != nil { - t.Fatal(err) - } - client, err := configread.NewClient(server.URL, localTestDispatcherID, "test-secret", server.Client()) - if err != nil { - t.Fatal(err) - } - broker := &fakeQueueBroker{} - queues := newTaskQueueController(d, broker, nil) - runtime := &LocalV01Runtime{dispatcher: d, client: client, queues: queues} - if err := runtime.refreshDiscoveryState(context.Background()); err != nil { - t.Fatal(err) - } - assignment, err := st.LocalTaskAssignment(localTestDispatcherID, localTestTaskID) - if err != nil { - t.Fatal(err) - } - if err := queues.StartTask(context.Background(), assignment); err != nil { - t.Fatal(err) - } - if err := runtime.refreshDiscoveryState(context.Background()); err != nil { - t.Fatal(err) - } - if err := runtime.reconcileTaskQueues(context.Background()); err != nil { - t.Fatal(err) - } - cursor, exists, err := st.LocalTaskDiscoveryCursor(localTestDispatcherID) - if err != nil || !exists || cursor != "opaque-watermark-002" || calls.Load() != 3 { - t.Fatalf("recovered cursor=%q exists=%v requests=%d err=%v", cursor, exists, calls.Load(), err) - } - assignment, err = st.LocalTaskAssignment(localTestDispatcherID, localTestTaskID) - if err != nil || !assignment.Removed || assignment.AdmissionState != "removed" { - t.Fatalf("retired assignment=%+v err=%v", assignment, err) - } - select { - case <-broker.consumers[0].done: - default: - t.Fatal("removed task remained subscribed") + if err := st.DB().QueryRow(`SELECT ready FROM local_v03_task_discovery_state WHERE dispatcher_id=?`, localTestDispatcherID).Scan(&ready); err != nil || ready != 1 { + t.Fatalf("discovery gate after fresh empty page=%d err=%v", ready, err) } } diff --git a/internal/store/local_discovery_helpers_test.go b/internal/store/local_discovery_helpers_test.go new file mode 100644 index 0000000..cd2e084 --- /dev/null +++ b/internal/store/local_discovery_helpers_test.go @@ -0,0 +1,96 @@ +package store + +import ( + "encoding/json" + "path/filepath" + "strconv" + "testing" + "time" +) + +func openLocalDiscoveryTestStore(t *testing.T) *Store { + t.Helper() + st, err := Open(filepath.Join(t.TempDir(), "dispatcher.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = st.Close() }) + return st +} + +func seedLocalCommandAssignments(t *testing.T, st *Store, records ...LocalCommandRecord) { + t.Helper() + if len(records) == 0 { + return + } + dispatcherID := records[0].DispatcherID + tasks := make([]LocalDiscoveredTask, 0, len(records)) + seen := make(map[string]struct{}, len(records)) + for _, record := range records { + if record.DispatcherID != dispatcherID { + t.Fatalf("test commands use different Dispatcher IDs: %q and %q", dispatcherID, record.DispatcherID) + } + if _, exists := seen[record.TaskID]; exists { + continue + } + seen[record.TaskID] = struct{}{} + revision := int64(1) + if record.Admission != nil && record.Admission.Task.TaskRevision > 0 { + revision = record.Admission.Task.TaskRevision + } + tasks = append(tasks, localDiscoveredTask(dispatcherID, record.TaskID, record.TenantID, record.TenantKey, revision, "running")) + } + seedLocalEventTasks(t, st, dispatcherID, tasks...) +} + +func seedLocalDiscovery(t *testing.T, st *Store) { + t.Helper() + seedLocalEventTasks(t, st, "d-1", + localDiscoveredTask("d-1", "task-a", "tenant-a", "tenant-key-a", 1, "running"), + localDiscoveredTask("d-1", "task-b", "tenant-a", "tenant-key-a", 1, "running"), + ) +} + +func seedLocalEventTasks(t *testing.T, st *Store, dispatcherID string, tasks ...LocalDiscoveredTask) { + t.Helper() + cursor, exists, err := st.LocalTaskDiscoveryCursor(dispatcherID) + if err != nil { + t.Fatal(err) + } + if !exists { + cursor = "0" + } + eventID, err := strconv.ParseUint(cursor, 10, 64) + if err != nil || eventID == ^uint64(0) { + t.Fatalf("invalid test event cursor %q: %v", cursor, err) + } + next := strconv.FormatUint(eventID+1, 10) + wire := make([]map[string]any, 0, len(tasks)) + for _, task := range tasks { + wire = append(wire, map[string]any{"task_id": task.TaskID, "tenant_id": task.TenantID, + "tenant_key": task.TenantKey, "status": task.Status, "task_revision": task.TaskRevision}) + } + body, err := json.Marshal(map[string]any{"schema_version": "task-discovery.v0.3-proposal", "dispatcher_id": dispatcherID, + "tasks": wire, "next_cursor": next}) + if err != nil { + t.Fatal(err) + } + now := time.Now().UTC() + if err := st.ApplyLocalTaskDiscoveryPage(LocalTaskDiscoveryPage{DispatcherID: dispatcherID, FromCursor: cursor, + NextCursor: next, Tasks: tasks, Body: body, ObservedAt: now}); err != nil { + t.Fatal(err) + } + empty, err := json.Marshal(map[string]any{"schema_version": "task-discovery.v0.3-proposal", "dispatcher_id": dispatcherID, + "tasks": []any{}, "next_cursor": next}) + if err != nil { + t.Fatal(err) + } + if err := st.ApplyLocalTaskDiscoveryPage(LocalTaskDiscoveryPage{DispatcherID: dispatcherID, FromCursor: next, + NextCursor: next, Body: empty, ObservedAt: now}); err != nil { + t.Fatal(err) + } +} + +func localDiscoveredTask(_ string, taskID, tenantID, tenantKey string, revision int64, status string) LocalDiscoveredTask { + return LocalDiscoveredTask{TaskID: taskID, TenantID: tenantID, TenantKey: tenantKey, TaskRevision: revision, Status: status} +} diff --git a/internal/store/local_origination.go b/internal/store/local_origination.go index da7f6c4..81bb002 100644 --- a/internal/store/local_origination.go +++ b/internal/store/local_origination.go @@ -47,7 +47,7 @@ func (s *Store) LoadLocalOrigination(dispatcherID, executionID string) (LocalOri JOIN local_v01_execution_configs c ON c.execution_id=t.execution_id JOIN reservations r ON r.execution_id=t.execution_id JOIN local_v01_task_assignments a ON a.dispatcher_id=? AND a.task_id=t.task_id - JOIN local_v02_task_discovery_state d ON d.dispatcher_id=a.dispatcher_id + JOIN local_v03_task_discovery_state d ON d.dispatcher_id=a.dispatcher_id WHERE t.execution_id=?`, dispatcherID, executionID).Scan(&candidate.TenantID, &candidate.TenantKey, &candidate.TaskID, &candidate.CommandID, &candidate.Callee, &taskStatus, &candidate.CommandBody, &inboxHash, &inboxStatus, &snapshotBody, &candidate.SnapshotSHA256, &reservationState, &admissionState, &removed, &ready) if errors.Is(err, sql.ErrNoRows) { @@ -75,7 +75,7 @@ func validateLocalOriginationCandidate(candidate *LocalOriginationCandidate, sna if err := json.Unmarshal(snapshotBody, &candidate.Snapshot); err != nil { return fmt.Errorf("decode bound execution snapshot: %w", err) } - if candidate.Snapshot.SchemaVersion != "execution-config-snapshot.v0.2" || candidate.Snapshot.DispatcherID != candidate.DispatcherID || + if candidate.Snapshot.SchemaVersion != "execution-config-snapshot.v0.3" || candidate.Snapshot.DispatcherID != candidate.DispatcherID || candidate.Snapshot.TaskID != candidate.TaskID || candidate.Snapshot.TenantID != candidate.TenantID || candidate.Snapshot.TenantKey != candidate.TenantKey || candidate.Snapshot.SelectedTrunkID == "" { return fmt.Errorf("%w: execution snapshot identity or selected trunk mismatch", ErrLocalOriginationNotAuthorized) @@ -104,7 +104,7 @@ func (s *Store) ClaimLocalOrigination(dispatcherID, executionID, agentID string, SELECT t.execution_id,?,'issued','',? FROM tasks t WHERE t.execution_id=? AND t.status='reserved' AND EXISTS (SELECT 1 FROM reservations r WHERE r.execution_id=t.execution_id AND r.state='held') AND EXISTS (SELECT 1 FROM local_v01_task_assignments a WHERE a.dispatcher_id=? AND a.task_id=t.task_id AND a.tenant_id=t.tenant_id AND a.tenant_key=t.tenant_key AND a.removed=0 AND a.admission_state='running') - AND EXISTS (SELECT 1 FROM local_v02_task_discovery_state d WHERE d.dispatcher_id=? AND d.ready=1) + AND EXISTS (SELECT 1 FROM local_v03_task_discovery_state d WHERE d.dispatcher_id=? AND d.ready=1) AND NOT EXISTS (SELECT 1 FROM controls c WHERE c.execution_id=t.execution_id AND c.action IN ('pause','drain','stop','hangup')) AND NOT EXISTS (SELECT 1 FROM local_v02_origination_decisions old WHERE old.execution_id=t.execution_id)`, dispatcherID, decidedAt.UTC().Format(time.RFC3339Nano), executionID, dispatcherID, dispatcherID) diff --git a/internal/store/local_v01.go b/internal/store/local_v01.go index 5679e50..b1d4499 100644 --- a/internal/store/local_v01.go +++ b/internal/store/local_v01.go @@ -49,7 +49,6 @@ type LocalExecutionConfigSnapshot struct { TaskRevision int64 `json:"task_revision"` QuotaRevision int64 `json:"quota_revision"` SIP json.RawMessage `json:"sip"` - Tasks json.RawMessage `json:"tasks"` Task json.RawMessage `json:"task"` TenantQuota json.RawMessage `json:"tenant_quota"` } @@ -191,9 +190,9 @@ func validateLocalAdmission(record LocalCommandRecord) ([]byte, error) { if len(admission.QuotaScopes) != 2 || admission.QuotaScopes[0] != LocalTenantQuotaScope(admission.DispatcherID, task.TenantID) || admission.QuotaScopes[1] != LocalTaskQuotaScope(admission.DispatcherID, task.TenantID, task.TaskID) { return nil, errors.New("local call admission quota scopes are invalid") } - if snapshot.SchemaVersion != "execution-config-snapshot.v0.2" || snapshot.SelectedTrunkID == "" || snapshot.DispatcherID != admission.DispatcherID || snapshot.TenantID != task.TenantID || snapshot.TenantKey != task.TenantKey || snapshot.TaskID != task.TaskID || + if snapshot.SchemaVersion != "execution-config-snapshot.v0.3" || snapshot.SelectedTrunkID == "" || snapshot.DispatcherID != admission.DispatcherID || snapshot.TenantID != task.TenantID || snapshot.TenantKey != task.TenantKey || snapshot.TaskID != task.TaskID || snapshot.TaskRevision != task.TaskRevision || snapshot.SIPRevision <= 0 || snapshot.QuotaRevision <= 0 || - !validLocalJSONResponse(snapshot.SIP) || !validLocalJSONResponse(snapshot.Tasks) || !validLocalJSONResponse(snapshot.Task) || !validLocalJSONResponse(snapshot.TenantQuota) { + !validLocalJSONResponse(snapshot.SIP) || !validLocalJSONResponse(snapshot.Task) || !validLocalJSONResponse(snapshot.TenantQuota) { return nil, errors.New("local call admission config snapshot identity or contents are invalid") } body, err := json.Marshal(snapshot) diff --git a/internal/store/local_v01_admission_test.go b/internal/store/local_v01_admission_test.go index a9caaa6..59e8214 100644 --- a/internal/store/local_v01_admission_test.go +++ b/internal/store/local_v01_admission_test.go @@ -171,10 +171,10 @@ func localAdmissionCommand(commandID, executionID, taskID string, bundle LocalCo responses[snapshot.Resource] = append(json.RawMessage(nil), snapshot.Body...) } config := LocalExecutionConfigSnapshot{ - SchemaVersion: "execution-config-snapshot.v0.2", DispatcherID: dispatcherID, + SchemaVersion: "execution-config-snapshot.v0.3", DispatcherID: dispatcherID, TenantID: tenantID, TenantKey: tenantKey, TaskID: taskID, SelectedTrunkID: "mock-trunk", SIPRevision: bundle.Snapshots[0].Revision, TaskRevision: 1, QuotaRevision: 1, - SIP: responses["sip"], Tasks: responses["tasks"], Task: responses["task"], TenantQuota: responses["tenant_quota"], + SIP: responses["sip"], Task: responses["task"], TenantQuota: responses["tenant_quota"], } task := Task{ ExecutionID: executionID, TenantKey: tenantKey, TenantID: tenantID, TaskID: taskID, diff --git a/internal/store/local_v01_config.go b/internal/store/local_v01_config.go index 20c474c..1bed777 100644 --- a/internal/store/local_v01_config.go +++ b/internal/store/local_v01_config.go @@ -54,10 +54,10 @@ func (s *Store) SaveLocalConfigBundle(bundle LocalConfigBundle) error { if bundle.DispatcherID == "" || bundle.TenantID == "" || bundle.TenantKey == "" || bundle.TaskID == "" || bundle.TenantQuotaLimit < 0 || bundle.TaskQuotaLimit < 0 || bundle.TenantQuotaScope != LocalTenantQuotaScope(bundle.DispatcherID, bundle.TenantID) || - bundle.TaskQuotaScope != LocalTaskQuotaScope(bundle.DispatcherID, bundle.TenantID, bundle.TaskID) || len(bundle.Snapshots) != 4 { + bundle.TaskQuotaScope != LocalTaskQuotaScope(bundle.DispatcherID, bundle.TenantID, bundle.TaskID) || len(bundle.Snapshots) != 3 { return errors.New("incomplete local configuration bundle") } - seen := make(map[string]bool, 4) + seen := make(map[string]bool, 3) for _, snapshot := range bundle.Snapshots { if seen[snapshot.Resource] || !json.Valid(snapshot.Body) || snapshot.Version == "" || snapshot.FetchedAt.IsZero() || !snapshot.ExpiresAt.After(snapshot.FetchedAt) { @@ -65,7 +65,7 @@ func (s *Store) SaveLocalConfigBundle(bundle LocalConfigBundle) error { } seen[snapshot.Resource] = true switch snapshot.Resource { - case "sip", "tasks": + case "sip": if snapshot.TenantID != "" || snapshot.TaskID != "" { return errors.New("shared configuration snapshot has task identity") } @@ -83,11 +83,8 @@ func (s *Store) SaveLocalConfigBundle(bundle LocalConfigBundle) error { if snapshot.Resource == "sip" && snapshot.Revision <= 0 { return errors.New("SIP configuration revision must be positive") } - if snapshot.Resource == "tasks" && snapshot.Revision != 0 { - return errors.New("task discovery uses an opaque version, not a numeric revision") - } } - for _, resource := range []string{"sip", "tasks", "task", "tenant_quota"} { + for _, resource := range []string{"sip", "task", "tenant_quota"} { if !seen[resource] { return fmt.Errorf("missing local configuration resource %q", resource) } @@ -112,17 +109,11 @@ func (s *Store) SaveLocalConfigBundle(bundle LocalConfigBundle) error { WHERE dispatcher_id=? AND resource=? AND tenant_id=? AND task_id=?`, bundle.DispatcherID, snapshot.Resource, snapshot.TenantID, snapshot.TaskID).Scan(&oldRevision, &oldVersion, &oldDigest) if err == nil { - if snapshot.Resource == "tasks" { - if snapshot.Version == oldVersion && digestHex != oldDigest { - return ErrConfigRevisionConflict - } - } else { - if snapshot.Revision < oldRevision { - return ErrConfigRevisionRollback - } - if snapshot.Revision == oldRevision && (snapshot.Version != oldVersion || digestHex != oldDigest) { - return ErrConfigRevisionConflict - } + if snapshot.Revision < oldRevision { + return ErrConfigRevisionRollback + } + if snapshot.Revision == oldRevision && (snapshot.Version != oldVersion || digestHex != oldDigest) { + return ErrConfigRevisionConflict } } else if !errors.Is(err, sql.ErrNoRows) { return err diff --git a/internal/store/local_v01_config_test.go b/internal/store/local_v01_config_test.go index f95ac48..019b08c 100644 --- a/internal/store/local_v01_config_test.go +++ b/internal/store/local_v01_config_test.go @@ -42,7 +42,7 @@ func TestSaveLocalConfigBundlePersistsFullResponsesAndSharedQuota(t *testing.T) if limit != 1 || reserved != 1 || unknown != 1 { t.Fatalf("tenant quota after downgrade = limit:%d reserved:%d unknown:%d", limit, reserved, unknown) } - if snapshot, err := st.LocalConfigSnapshot("d-1", "tenant_quota", "tenant-a", ""); err != nil || string(snapshot.Body) != string(second.Snapshots[3].Body) { + if snapshot, err := st.LocalConfigSnapshot("d-1", "tenant_quota", "tenant-a", ""); err != nil || string(snapshot.Body) != string(second.Snapshots[2].Body) { t.Fatalf("shared tenant quota body=%q err=%v", snapshot.Body, err) } } @@ -70,7 +70,7 @@ func TestSaveLocalConfigBundleRejectsRevisionAndTenantBindingConflicts(t *testin } } changedSameRevision := localConfigBundle("d-1", "tenant-a", "tenant-key-a", "task-a", 2, 2, 2, 3, 3, now.Add(time.Second)) - changedSameRevision.Snapshots[2].Body = []byte(`{"resource":"task","revision":2,"different":true}`) + changedSameRevision.Snapshots[1].Body = []byte(`{"resource":"task","revision":2,"different":true}`) if err := st.SaveLocalConfigBundle(changedSameRevision); !errors.Is(err, ErrConfigRevisionConflict) { t.Fatalf("same revision content conflict error = %v, want ErrConfigRevisionConflict", err) } @@ -89,7 +89,6 @@ func localConfigBundle(dispatcherID, tenantID, tenantKey, taskID string, sipRevi TenantQuotaLimit: tenantLimit, TaskQuotaLimit: taskLimit, Snapshots: []LocalConfigSnapshotRecord{ {Resource: "sip", Revision: sipRevision, Version: fmt.Sprint(sipRevision), FetchedAt: fetchedAt, ExpiresAt: expires, Body: []byte(fmt.Sprintf(`{"resource":"sip","revision":%d}`, sipRevision))}, - {Resource: "tasks", Version: fmt.Sprintf("snapshot-%s", taskID), FetchedAt: fetchedAt, ExpiresAt: expires, Body: []byte(fmt.Sprintf(`{"resource":"tasks","snapshot_id":"snapshot-%s"}`, taskID))}, {Resource: "task", TenantID: tenantID, TaskID: taskID, Revision: taskRevision, Version: fmt.Sprint(taskRevision), FetchedAt: fetchedAt, ExpiresAt: expires, Body: []byte(fmt.Sprintf(`{"resource":"task","revision":%d,"task_id":%q}`, taskRevision, taskID))}, {Resource: "tenant_quota", TenantID: tenantID, Revision: quotaRevision, Version: fmt.Sprint(quotaRevision), FetchedAt: fetchedAt, ExpiresAt: expires, Body: []byte(fmt.Sprintf(`{"resource":"tenant_quota","revision":%d,"tenant_key":%q}`, quotaRevision, tenantKey))}, }, diff --git a/internal/store/local_v01_discovery.go b/internal/store/local_v01_discovery.go index ae9a6ab..02661ad 100644 --- a/internal/store/local_v01_discovery.go +++ b/internal/store/local_v01_discovery.go @@ -1,10 +1,7 @@ package store import ( - "crypto/sha256" "database/sql" - "encoding/hex" - "encoding/json" "errors" "fmt" "time" @@ -23,17 +20,6 @@ var ( ErrLocalTaskStopped = errors.New("task admission is terminal") ) -type LocalTaskDiscovery struct { - DispatcherID string - Mode string - Cursor string - FromCursor string - NextCursor string - Body []byte - Tasks []LocalDiscoveredTask - Changes []LocalTaskDiscoveryChange -} - type LocalDiscoveredTask struct { TaskID string TenantID string @@ -42,15 +28,6 @@ type LocalDiscoveredTask struct { TaskRevision int64 } -type LocalTaskDiscoveryChange struct { - Cursor string - Operation string - TaskID string - TenantID string - TenantKey string - Task LocalDiscoveredTask -} - type LocalTaskQueue struct { Exchange string RoutingKey string @@ -59,139 +36,17 @@ type LocalTaskQueue struct { } type LocalTaskAssignment struct { - DispatcherID string - TaskID string - TenantID string - TenantKey string - TaskRevision int64 - Status string - AdmissionState string - Removed bool - Queue LocalTaskQueue - UpdatedAt string -} - -// ApplyLocalTaskDiscovery commits the complete response, assignments, and cursor -// together. An invalid response cannot advance durable discovery state. -func (s *Store) ApplyLocalTaskDiscovery(discovery LocalTaskDiscovery) error { - if err := validateLocalTaskDiscovery(discovery); err != nil { - return err - } - now := s.now().UTC().Format(time.RFC3339Nano) - s.mu.Lock() - defer s.mu.Unlock() - tx, err := s.db.Begin() - if err != nil { - return err - } - defer tx.Rollback() - - cursor := discovery.Cursor - if discovery.Mode == "changes" { - var currentCursor string - if err := tx.QueryRow(`SELECT cursor FROM local_v02_task_discovery_state WHERE dispatcher_id=?`, discovery.DispatcherID).Scan(¤tCursor); err != nil { - if errors.Is(err, sql.ErrNoRows) { - return ErrLocalDiscoveryCursorMismatch - } - return err - } - if currentCursor != discovery.FromCursor { - return ErrLocalDiscoveryCursorMismatch - } - cursor = discovery.NextCursor - } - - if discovery.Mode == "snapshot" { - seen := make(map[string]struct{}, len(discovery.Tasks)) - for _, task := range discovery.Tasks { - if err := bindLocalTenant(tx, discovery.DispatcherID, task.TenantID, task.TenantKey, s.now()); err != nil { - return err - } - if err := applyLocalDiscoveredTask(tx, discovery.DispatcherID, task, true, now); err != nil { - return err - } - seen[task.TaskID] = struct{}{} - } - if err := markLocalTasksAbsent(tx, discovery.DispatcherID, seen, now); err != nil { - return err - } - } else { - for _, change := range discovery.Changes { - if err := applyLocalDiscoveryChange(tx, discovery.DispatcherID, change, now, s.now()); err != nil { - return err - } - } - } - var activeAssignments int - if err := tx.QueryRow(`SELECT COUNT(*) FROM local_v01_task_assignments WHERE dispatcher_id=? AND removed=0`, discovery.DispatcherID).Scan(&activeAssignments); err != nil { - return err - } - if activeAssignments > 256 { - return ErrLocalDiscoveryTaskLimit - } - digest := sha256.Sum256(discovery.Body) - if _, err := tx.Exec(`INSERT INTO local_v02_task_discovery_response(dispatcher_id,content_sha256,body,stored_at) - VALUES(?,?,?,?) ON CONFLICT(dispatcher_id) DO UPDATE SET content_sha256=excluded.content_sha256, - body=excluded.body,stored_at=excluded.stored_at`, discovery.DispatcherID, hex.EncodeToString(digest[:]), discovery.Body, now); err != nil { - return err - } - if _, err := tx.Exec(`INSERT INTO local_v02_task_discovery_state(dispatcher_id,cursor,last_mode,ready,updated_at) - VALUES(?,?,?,1,?) ON CONFLICT(dispatcher_id) DO UPDATE SET cursor=excluded.cursor, - last_mode=excluded.last_mode,ready=1,updated_at=excluded.updated_at`, discovery.DispatcherID, cursor, discovery.Mode, now); err != nil { - return err - } - return tx.Commit() -} - -func validateLocalTaskDiscovery(discovery LocalTaskDiscovery) error { - if discovery.DispatcherID == "" || len(discovery.Body) == 0 || !json.Valid(discovery.Body) { - return errors.New("task discovery requires a Dispatcher ID and one valid response body") - } - switch discovery.Mode { - case "snapshot": - if discovery.Cursor == "" || discovery.FromCursor != "" || discovery.NextCursor != "" || len(discovery.Changes) != 0 || len(discovery.Tasks) > 256 { - return errors.New("invalid task-discovery snapshot window") - } - seen := make(map[string]struct{}, len(discovery.Tasks)) - for _, task := range discovery.Tasks { - if err := validateLocalDiscoveredTask(discovery.DispatcherID, task); err != nil { - return err - } - if _, exists := seen[task.TaskID]; exists { - return fmt.Errorf("task discovery contains duplicate task ID %q", task.TaskID) - } - seen[task.TaskID] = struct{}{} - } - case "changes": - if discovery.FromCursor == "" || discovery.NextCursor == "" || discovery.Cursor != "" || len(discovery.Tasks) != 0 || len(discovery.Changes) > 256 || - (len(discovery.Changes) == 0 && discovery.NextCursor != discovery.FromCursor) || - (len(discovery.Changes) != 0 && discovery.NextCursor == discovery.FromCursor) { - return errors.New("invalid task-discovery change window") - } - for _, change := range discovery.Changes { - if change.Cursor == "" { - return errors.New("task-discovery change cursor is required") - } - switch change.Operation { - case "assigned", "updated": - if err := validateLocalDiscoveredTask(discovery.DispatcherID, change.Task); err != nil { - return err - } - if change.TaskID != "" && change.TaskID != change.Task.TaskID { - return errors.New("task-discovery change task ID conflicts with its task payload") - } - case "removed": - if !validLocalTaskID(change.TaskID) || change.TenantID == "" || change.TenantKey == "" || len([]byte(change.TenantKey)) > 196 { - return errors.New("invalid task-discovery removal tombstone") - } - default: - return fmt.Errorf("unsupported task-discovery operation %q", change.Operation) - } - } - default: - return fmt.Errorf("unsupported task-discovery mode %q", discovery.Mode) - } - return nil + DispatcherID string + TaskID string + TenantID string + TenantKey string + TaskRevision int64 + Status string + AdmissionState string + Removed bool + Queue LocalTaskQueue + UpdatedAt string + DiscoveryObservedAt time.Time } func validateLocalDiscoveredTask(dispatcherID string, task LocalDiscoveredTask) error { @@ -199,7 +54,7 @@ func validateLocalDiscoveredTask(dispatcherID string, task LocalDiscoveredTask) return errors.New("incomplete task-discovery assignment") } switch task.Status { - case "running", "paused", "stopped", "finished": + case "running", "paused", "stopped", "finished", "removed": default: return fmt.Errorf("unsupported discovered task status %q", task.Status) } @@ -224,7 +79,7 @@ func derivedLocalTaskQueue(dispatcherID, taskID string) LocalTaskQueue { QueueName: fmt.Sprintf("agent-call.d.%s.task.%s.v3", dispatcherID, taskID)} } -func applyLocalDiscoveredTask(tx *sql.Tx, dispatcherID string, task LocalDiscoveredTask, allowCreate bool, now string) error { +func applyLocalDiscoveredTask(tx *sql.Tx, dispatcherID string, task LocalDiscoveredTask, now string) error { queue := derivedLocalTaskQueue(dispatcherID, task.TaskID) var old LocalTaskAssignment var removed int @@ -233,12 +88,13 @@ func applyLocalDiscoveredTask(tx *sql.Tx, dispatcherID string, task LocalDiscove dispatcherID, task.TaskID).Scan(&old.DispatcherID, &old.TaskID, &old.TenantID, &old.TenantKey, &old.TaskRevision, &old.Status, &old.AdmissionState, &removed, &old.Queue.Exchange, &old.Queue.RoutingKey, &old.Queue.BindingKey, &old.Queue.QueueName, &old.UpdatedAt) if errors.Is(err, sql.ErrNoRows) { - if !allowCreate { - return ErrLocalDiscoveryTaskMissing + removed := 0 + if task.Status == "removed" { + removed = 1 } _, err = tx.Exec(`INSERT INTO local_v01_task_assignments(dispatcher_id,task_id,tenant_id,tenant_key,task_revision,saas_status,admission_state,removed, - queue_exchange,routing_key,binding_key,queue_name,updated_at) VALUES(?,?,?,?,?,?,?,0,?,?,?,?,?)`, - dispatcherID, task.TaskID, task.TenantID, task.TenantKey, task.TaskRevision, task.Status, task.Status, + queue_exchange,routing_key,binding_key,queue_name,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?)`, + dispatcherID, task.TaskID, task.TenantID, task.TenantKey, task.TaskRevision, task.Status, task.Status, removed, queue.Exchange, queue.RoutingKey, queue.BindingKey, queue.QueueName, now) return err } @@ -246,24 +102,31 @@ func applyLocalDiscoveredTask(tx *sql.Tx, dispatcherID string, task LocalDiscove return err } old.Removed = removed != 0 - if old.Removed { - return ErrLocalTaskRemoved - } if old.TenantID != task.TenantID || old.TenantKey != task.TenantKey { return ErrTenantBindingConflict } if task.TaskRevision < old.TaskRevision { return ErrConfigRevisionRollback } - if task.TaskRevision == old.TaskRevision { - if old.Status != task.Status || old.Queue != queue { - return ErrConfigRevisionConflict + if old.Removed { + if task.Status != "removed" { + return ErrLocalTaskRemoved } return nil } + if old.Queue != queue { + return ErrConfigRevisionConflict + } + if task.TaskRevision == old.TaskRevision && task.Status == old.Status { + return nil + } state := mergeLocalTaskState(old.AdmissionState, task.Status) - _, err = tx.Exec(`UPDATE local_v01_task_assignments SET task_revision=?,saas_status=?,admission_state=?,queue_exchange=?,routing_key=?,binding_key=?,queue_name=?,updated_at=? - WHERE dispatcher_id=? AND task_id=?`, task.TaskRevision, task.Status, state, queue.Exchange, queue.RoutingKey, + removed = 0 + if task.Status == "removed" { + state, removed = "removed", 1 + } + _, err = tx.Exec(`UPDATE local_v01_task_assignments SET task_revision=?,saas_status=?,admission_state=?,removed=?,queue_exchange=?,routing_key=?,binding_key=?,queue_name=?,updated_at=? + WHERE dispatcher_id=? AND task_id=?`, task.TaskRevision, task.Status, state, removed, queue.Exchange, queue.RoutingKey, queue.BindingKey, queue.QueueName, now, dispatcherID, task.TaskID) return err } @@ -288,94 +151,30 @@ func mergeLocalTaskState(current, incoming string) string { } } -func applyLocalDiscoveryChange(tx *sql.Tx, dispatcherID string, change LocalTaskDiscoveryChange, now string, createdAt time.Time) error { - switch change.Operation { - case "assigned": - if err := bindLocalTenant(tx, dispatcherID, change.Task.TenantID, change.Task.TenantKey, createdAt); err != nil { - return err - } - return applyLocalDiscoveredTask(tx, dispatcherID, change.Task, true, now) - case "updated": - if err := bindLocalTenant(tx, dispatcherID, change.Task.TenantID, change.Task.TenantKey, createdAt); err != nil { - return err - } - return applyLocalDiscoveredTask(tx, dispatcherID, change.Task, false, now) - case "removed": - if err := bindLocalTenant(tx, dispatcherID, change.TenantID, change.TenantKey, createdAt); err != nil { - return err - } - return removeLocalTaskTx(tx, dispatcherID, change, now) - default: - return fmt.Errorf("unsupported task-discovery operation %q", change.Operation) - } -} - -func removeLocalTaskTx(tx *sql.Tx, dispatcherID string, change LocalTaskDiscoveryChange, now string) error { - var tenantID, tenantKey string - err := tx.QueryRow(`SELECT tenant_id,tenant_key FROM local_v01_task_assignments WHERE dispatcher_id=? AND task_id=?`, dispatcherID, change.TaskID).Scan(&tenantID, &tenantKey) - if errors.Is(err, sql.ErrNoRows) { - queue := derivedLocalTaskQueue(dispatcherID, change.TaskID) - _, err = tx.Exec(`INSERT INTO local_v01_task_assignments(dispatcher_id,task_id,tenant_id,tenant_key,task_revision,saas_status,admission_state,removed, - queue_exchange,routing_key,binding_key,queue_name,updated_at) VALUES(?,?,?,?,0,'removed','removed',1,?,?,?,?,?)`, - dispatcherID, change.TaskID, change.TenantID, change.TenantKey, queue.Exchange, queue.RoutingKey, queue.BindingKey, queue.QueueName, now) - return err - } - if err != nil { - return err - } - if tenantID != change.TenantID || tenantKey != change.TenantKey { - return ErrTenantBindingConflict - } - _, err = tx.Exec(`UPDATE local_v01_task_assignments SET saas_status='removed',admission_state='removed',removed=1,updated_at=? - WHERE dispatcher_id=? AND task_id=?`, now, dispatcherID, change.TaskID) - return err -} - -func markLocalTasksAbsent(tx *sql.Tx, dispatcherID string, present map[string]struct{}, now string) error { - rows, err := tx.Query(`SELECT task_id FROM local_v01_task_assignments WHERE dispatcher_id=? AND removed=0`, dispatcherID) - if err != nil { - return err - } - var absent []string - for rows.Next() { - var taskID string - if err := rows.Scan(&taskID); err != nil { - _ = rows.Close() - return err - } - if _, exists := present[taskID]; !exists { - absent = append(absent, taskID) - } - } - if err := rows.Err(); err != nil { - _ = rows.Close() - return err - } - if err := rows.Close(); err != nil { - return err - } - for _, taskID := range absent { - if _, err := tx.Exec(`UPDATE local_v01_task_assignments SET saas_status='removed',admission_state='removed',removed=1,updated_at=? WHERE dispatcher_id=? AND task_id=?`, now, dispatcherID, taskID); err != nil { - return err - } - } - return nil -} - // CloseLocalTaskDiscoveryAdmission rejects new work without losing the cursor // or changing an authoritative task pause/stop state. func (s *Store) CloseLocalTaskDiscoveryAdmission(dispatcherID string) error { s.mu.Lock() defer s.mu.Unlock() - _, err := s.db.Exec(`UPDATE local_v02_task_discovery_state SET ready=0 WHERE dispatcher_id=?`, dispatcherID) + _, err := s.db.Exec(`UPDATE local_v03_task_discovery_state SET ready=0 WHERE dispatcher_id=?`, dispatcherID) return err } func (s *Store) LocalTaskDiscoveryCursor(dispatcherID string) (cursor string, exists bool, err error) { s.mu.Lock() defer s.mu.Unlock() - err = s.db.QueryRow(`SELECT cursor FROM local_v02_task_discovery_state WHERE dispatcher_id=?`, dispatcherID).Scan(&cursor) + err = s.db.QueryRow(`SELECT cursor FROM local_v03_task_discovery_state WHERE dispatcher_id=?`, dispatcherID).Scan(&cursor) if errors.Is(err, sql.ErrNoRows) { + var legacy, assignments int + if err := s.db.QueryRow(`SELECT COUNT(*) FROM local_v02_task_discovery_state WHERE dispatcher_id=?`, dispatcherID).Scan(&legacy); err != nil { + return "", false, err + } + if err := s.db.QueryRow(`SELECT COUNT(*) FROM local_v01_task_assignments WHERE dispatcher_id=?`, dispatcherID).Scan(&assignments); err != nil { + return "", false, err + } + if legacy != 0 || assignments != 0 { + return "", false, fmt.Errorf("%w: existing discovery state requires controlled drain before v0.3", ErrLocalDiscoveryUnavailable) + } return "", false, nil } if err != nil { @@ -429,7 +228,7 @@ func scanLocalTaskAssignment(row interface{ Scan(...any) error }) (LocalTaskAssi func requireLocalTaskRunningTx(tx *sql.Tx, dispatcherID, tenantID, tenantKey, taskID string) error { var ready int - if err := tx.QueryRow(`SELECT ready FROM local_v02_task_discovery_state WHERE dispatcher_id=?`, dispatcherID).Scan(&ready); err != nil { + if err := tx.QueryRow(`SELECT ready FROM local_v03_task_discovery_state WHERE dispatcher_id=?`, dispatcherID).Scan(&ready); err != nil { if errors.Is(err, sql.ErrNoRows) { return ErrLocalDiscoveryUnavailable } diff --git a/internal/store/local_v01_discovery_test.go b/internal/store/local_v01_discovery_test.go deleted file mode 100644 index 4d74b82..0000000 --- a/internal/store/local_v01_discovery_test.go +++ /dev/null @@ -1,292 +0,0 @@ -package store - -import ( - "crypto/sha256" - "database/sql" - "encoding/hex" - "errors" - "fmt" - "path/filepath" - "testing" - "time" - - "git.ipao.vip/rogee/go-sip/internal/mq" -) - -func TestApplyLocalTaskDiscoveryPersistsSingleResponseAndDerivedQueue(t *testing.T) { - st := openLocalDiscoveryTestStore(t) - body := []byte(`{"cursor":"opaque-cursor-1","tasks":[{"task_id":"task-a"}]}`) - if err := st.ApplyLocalTaskDiscovery(localSnapshotWindow("d-1", "opaque-cursor-1", body, - localDiscoveredTask("d-1", "task-a", "tenant-a", "tenant-key-a", 1, "running"), - localDiscoveredTask("d-1", "task-b", "tenant-a", "tenant-key-a", 4, "paused"), - )); err != nil { - t.Fatal(err) - } - cursor, exists, err := st.LocalTaskDiscoveryCursor("d-1") - if err != nil || !exists || cursor != "opaque-cursor-1" { - t.Fatalf("cursor=%q exists=%v err=%v", cursor, exists, err) - } - assignments, err := st.LocalTaskAssignments("d-1") - if err != nil || len(assignments) != 2 { - t.Fatalf("assignments=%+v err=%v", assignments, err) - } - if assignments[0].TaskID != "task-a" || assignments[0].AdmissionState != "running" || assignments[1].AdmissionState != "paused" { - t.Fatalf("initial task states = %+v", assignments) - } - if assignments[0].Queue.Exchange != mq.CommandsExchangeV3 || assignments[0].Queue.RoutingKey != "d.d-1.task.task-a.in" || assignments[0].Queue.BindingKey != "d.d-1.task.task-a.in" || assignments[0].Queue.QueueName != "agent-call.d.d-1.task.task-a.v3" { - t.Fatalf("derived queue = %+v", assignments[0].Queue) - } - var saved []byte - var digest string - if err := st.db.QueryRow(`SELECT body,content_sha256 FROM local_v02_task_discovery_response WHERE dispatcher_id=?`, "d-1").Scan(&saved, &digest); err != nil { - t.Fatal(err) - } - hash := sha256.Sum256(body) - if string(saved) != string(body) || digest != hex.EncodeToString(hash[:]) { - t.Fatalf("stored discovery response mismatch: %s %s", saved, digest) - } -} - -func TestApplyLocalTaskDiscoveryChangesAndCursorAdvanceAreAtomic(t *testing.T) { - st := openLocalDiscoveryTestStore(t) - seedLocalDiscovery(t, st) - changes := LocalTaskDiscovery{ - DispatcherID: "d-1", Mode: "changes", FromCursor: "cursor-1", NextCursor: "cursor-3", Body: []byte(`{"changes":[{"operation":"updated"},{"operation":"removed"}]}`), - Changes: []LocalTaskDiscoveryChange{ - {Cursor: "cursor-2", Operation: "updated", Task: localDiscoveredTask("d-1", "task-a", "tenant-a", "tenant-key-a", 2, "paused")}, - {Cursor: "cursor-3", Operation: "removed", TaskID: "task-b", TenantID: "tenant-a", TenantKey: "tenant-key-a"}, - }, - } - if err := st.ApplyLocalTaskDiscovery(changes); err != nil { - t.Fatal(err) - } - cursor, exists, err := st.LocalTaskDiscoveryCursor("d-1") - if err != nil || !exists || cursor != "cursor-3" { - t.Fatalf("cursor=%q exists=%v err=%v", cursor, exists, err) - } - assignments, err := st.LocalTaskAssignments("d-1") - if err != nil || len(assignments) != 2 { - t.Fatalf("assignments=%+v err=%v", assignments, err) - } - if assignments[0].TaskRevision != 2 || assignments[0].Status != "paused" || assignments[0].AdmissionState != "paused" || !assignments[1].Removed || assignments[1].AdmissionState != "removed" { - t.Fatalf("change application = %+v", assignments) - } - var stored []byte - if err := st.db.QueryRow(`SELECT body FROM local_v02_task_discovery_response WHERE dispatcher_id=?`, "d-1").Scan(&stored); err != nil || string(stored) != string(changes.Body) { - t.Fatalf("stored response=%s err=%v", stored, err) - } -} - -func TestApplyLocalTaskDiscoveryNoChangeKeepsCursor(t *testing.T) { - st := openLocalDiscoveryTestStore(t) - seedLocalDiscovery(t, st) - if err := st.ApplyLocalTaskDiscovery(LocalTaskDiscovery{DispatcherID: "d-1", Mode: "changes", FromCursor: "cursor-1", NextCursor: "cursor-1", Body: []byte(`{"changes":[]}`)}); err != nil { - t.Fatal(err) - } - cursor, exists, err := st.LocalTaskDiscoveryCursor("d-1") - if err != nil || !exists || cursor != "cursor-1" { - t.Fatalf("empty change cursor=%q exists=%v err=%v", cursor, exists, err) - } -} - -func TestLocalTaskDiscoveryFailureClosesAdmissionUntilFreshResponse(t *testing.T) { - st := openLocalDiscoveryTestStore(t) - seedLocalDiscovery(t, st) - if err := st.CloseLocalTaskDiscoveryAdmission("d-1"); err != nil { - t.Fatal(err) - } - now := time.Date(2026, 9, 21, 2, 0, 0, 0, time.UTC) - bundle := localConfigBundle("d-1", "tenant-a", "tenant-key-a", "task-a", 1, 1, 1, 1, 1, now) - record := localAdmissionCommand("command-a", "execution-a", "task-a", bundle, "d-1", "tenant-a", "tenant-key-a") - record.DispatcherID = "d-1" - if _, err := st.PersistLocalCommand(record); !errors.Is(err, ErrLocalDiscoveryUnavailable) { - t.Fatalf("admission despite failed discovery: %v", err) - } - cursor, exists, err := st.LocalTaskDiscoveryCursor("d-1") - if err != nil || !exists || cursor != "cursor-1" { - t.Fatalf("lost durable cursor: %q exists=%v err=%v", cursor, exists, err) - } - if err := st.ApplyLocalTaskDiscovery(LocalTaskDiscovery{DispatcherID: "d-1", Mode: "changes", FromCursor: "cursor-1", NextCursor: "cursor-1", Body: []byte(`{"changes":[]}`)}); err != nil { - t.Fatal(err) - } - if _, err := st.PersistLocalCommand(record); !errors.Is(err, ErrNoCapacity) { - t.Fatalf("fresh discovery should pass the discovery gate and reach the quota check, got %v", err) - } -} - -func TestApplyLocalTaskDiscoveryRejectsCursorMismatchWithoutAdvancing(t *testing.T) { - st := openLocalDiscoveryTestStore(t) - seedLocalDiscovery(t, st) - bad := LocalTaskDiscovery{DispatcherID: "d-1", Mode: "changes", FromCursor: "wrong-cursor", NextCursor: "cursor-2", Body: []byte(`{"changes":[{}]}`), Changes: []LocalTaskDiscoveryChange{{Cursor: "cursor-2", Operation: "updated", Task: localDiscoveredTask("d-1", "task-a", "tenant-a", "tenant-key-a", 2, "paused")}}} - if err := st.ApplyLocalTaskDiscovery(bad); !errors.Is(err, ErrLocalDiscoveryCursorMismatch) { - t.Fatalf("cursor mismatch error=%v", err) - } - cursor, _, err := st.LocalTaskDiscoveryCursor("d-1") - if err != nil || cursor != "cursor-1" { - t.Fatalf("cursor after rejected response=%q err=%v", cursor, err) - } -} - -func TestApplyLocalTaskDiscoveryRollsBackEarlierChangesOnLaterConflict(t *testing.T) { - st := openLocalDiscoveryTestStore(t) - seedLocalDiscovery(t, st) - bad := LocalTaskDiscovery{DispatcherID: "d-1", Mode: "changes", FromCursor: "cursor-1", NextCursor: "cursor-3", Body: []byte(`{"changes":[{},{}]}`), Changes: []LocalTaskDiscoveryChange{ - {Cursor: "cursor-2", Operation: "updated", Task: localDiscoveredTask("d-1", "task-a", "tenant-a", "tenant-key-a", 2, "paused")}, - {Cursor: "cursor-3", Operation: "updated", Task: localDiscoveredTask("d-1", "task-a", "tenant-a", "tenant-key-a", 1, "running")}, - }} - if err := st.ApplyLocalTaskDiscovery(bad); !errors.Is(err, ErrConfigRevisionRollback) { - t.Fatalf("revision rollback error=%v", err) - } - cursor, _, err := st.LocalTaskDiscoveryCursor("d-1") - if err != nil || cursor != "cursor-1" { - t.Fatalf("cursor after rollback=%q err=%v", cursor, err) - } - task, err := st.LocalTaskAssignment("d-1", "task-a") - if err != nil || task.TaskRevision != 1 || task.Status != "running" || task.AdmissionState != "running" { - t.Fatalf("task after rollback=%+v err=%v", task, err) - } -} - -func TestApplyLocalTaskDiscoveryRejectsMoreThan256AssignedTasksWithoutAdvancing(t *testing.T) { - st := openLocalDiscoveryTestStore(t) - tasks := make([]LocalDiscoveredTask, 256) - for i := range tasks { - tasks[i] = localDiscoveredTask("d-1", fmt.Sprintf("task-%03d", i), "tenant-a", "tenant-key-a", 1, "running") - } - if err := st.ApplyLocalTaskDiscovery(localSnapshotWindow("d-1", "cursor-1", []byte(`{"tasks":[]}`), tasks...)); err != nil { - t.Fatal(err) - } - change := LocalTaskDiscovery{DispatcherID: "d-1", Mode: "changes", FromCursor: "cursor-1", NextCursor: "cursor-2", Body: []byte(`{"changes":[{}]}`), Changes: []LocalTaskDiscoveryChange{ - {Cursor: "cursor-2", Operation: "assigned", Task: localDiscoveredTask("d-1", "task-256", "tenant-a", "tenant-key-a", 1, "running")}, - }} - if err := st.ApplyLocalTaskDiscovery(change); !errors.Is(err, ErrLocalDiscoveryTaskLimit) { - t.Fatalf("task overflow error=%v", err) - } - cursor, exists, err := st.LocalTaskDiscoveryCursor("d-1") - if err != nil || !exists || cursor != "cursor-1" { - t.Fatalf("task overflow advanced cursor: %q exists=%v err=%v", cursor, exists, err) - } - if _, err := st.LocalTaskAssignment("d-1", "task-256"); !errors.Is(err, sql.ErrNoRows) { - t.Fatalf("task overflow persisted partial assignment: %v", err) - } -} - -func TestApplyLocalTaskDiscoverySnapshotMarksOmittedTasksRemoved(t *testing.T) { - st := openLocalDiscoveryTestStore(t) - seedLocalDiscovery(t, st) - if err := st.ApplyLocalTaskDiscovery(localSnapshotWindow("d-1", "cursor-2", []byte(`{"tasks":[{}]}`), - localDiscoveredTask("d-1", "task-a", "tenant-a", "tenant-key-a", 1, "running"), - )); err != nil { - t.Fatal(err) - } - task, err := st.LocalTaskAssignment("d-1", "task-b") - if err != nil || !task.Removed || task.AdmissionState != "removed" { - t.Fatalf("omitted task=%+v err=%v", task, err) - } -} - -func TestApplyLocalTaskDiscoveryRejectsTenantBindingConflict(t *testing.T) { - st := openLocalDiscoveryTestStore(t) - bad := localSnapshotWindow("d-1", "cursor-1", []byte(`{"tasks":[{},{}]}`), - localDiscoveredTask("d-1", "task-a", "tenant-a", "tenant-key-a", 1, "running"), - localDiscoveredTask("d-1", "task-b", "tenant-a", "tenant-key-b", 1, "running"), - ) - if err := st.ApplyLocalTaskDiscovery(bad); !errors.Is(err, ErrTenantBindingConflict) { - t.Fatalf("tenant binding error=%v", err) - } - if _, exists, err := st.LocalTaskDiscoveryCursor("d-1"); err != nil || exists { - t.Fatalf("partial snapshot cursor persisted=%v err=%v", exists, err) - } -} - -func TestPersistLocalCommandRequiresRunningDiscoveredTask(t *testing.T) { - for _, tc := range []struct { - name string - discovered bool - status string - wantErr error - }{ - {name: "paused", discovered: true, status: "paused", wantErr: ErrLocalTaskPaused}, - {name: "stopped", discovered: true, status: "stopped", wantErr: ErrLocalTaskStopped}, - {name: "no-discovery", wantErr: ErrLocalDiscoveryUnavailable}, - } { - t.Run(tc.name, func(t *testing.T) { - st := openLocalDiscoveryTestStore(t) - if tc.discovered { - if err := st.ApplyLocalTaskDiscovery(localSnapshotWindow("d-1", "cursor-1", []byte(`{"tasks":[{}]}`), - localDiscoveredTask("d-1", "task-a", "tenant-a", "tenant-key-a", 1, tc.status), - )); err != nil { - t.Fatal(err) - } - } - now := time.Date(2026, 9, 21, 2, 0, 0, 0, time.UTC) - bundle := localConfigBundle("d-1", "tenant-a", "tenant-key-a", "task-a", 1, 1, 1, 1, 1, now) - record := localAdmissionCommand("command-a", "execution-a", "task-a", bundle, "d-1", "tenant-a", "tenant-key-a") - record.DispatcherID = "d-1" - if _, err := st.PersistLocalCommand(record); !errors.Is(err, tc.wantErr) { - t.Fatalf("PersistLocalCommand error=%v, want %v", err, tc.wantErr) - } - for _, table := range []string{"inbox", "outbox", "tasks", "reservations"} { - var count int - if err := st.db.QueryRow(`SELECT COUNT(*) FROM ` + table).Scan(&count); err != nil || count != 0 { - t.Fatalf("table %s rows=%d err=%v after blocked admission", table, count, err) - } - } - }) - } -} - -func openLocalDiscoveryTestStore(t *testing.T) *Store { - t.Helper() - st, err := Open(filepath.Join(t.TempDir(), "dispatcher.db")) - if err != nil { - t.Fatal(err) - } - t.Cleanup(func() { _ = st.Close() }) - return st -} - -func seedLocalCommandAssignments(t *testing.T, st *Store, records ...LocalCommandRecord) { - t.Helper() - if len(records) == 0 { - return - } - dispatcherID := records[0].DispatcherID - tasks := make([]LocalDiscoveredTask, 0, len(records)) - seen := make(map[string]struct{}, len(records)) - for _, record := range records { - if record.DispatcherID != dispatcherID { - t.Fatalf("test commands use different Dispatcher IDs: %q and %q", dispatcherID, record.DispatcherID) - } - if _, exists := seen[record.TaskID]; exists { - continue - } - seen[record.TaskID] = struct{}{} - revision := int64(1) - if record.Admission != nil && record.Admission.Task.TaskRevision > 0 { - revision = record.Admission.Task.TaskRevision - } - tasks = append(tasks, localDiscoveredTask(dispatcherID, record.TaskID, record.TenantID, record.TenantKey, revision, "running")) - } - if err := st.ApplyLocalTaskDiscovery(localSnapshotWindow(dispatcherID, "test-cursor", []byte(`{"tasks":[]}`), tasks...)); err != nil { - t.Fatal(err) - } -} - -func seedLocalDiscovery(t *testing.T, st *Store) { - t.Helper() - if err := st.ApplyLocalTaskDiscovery(localSnapshotWindow("d-1", "cursor-1", []byte(`{"tasks":[{},{}]}`), - localDiscoveredTask("d-1", "task-a", "tenant-a", "tenant-key-a", 1, "running"), - localDiscoveredTask("d-1", "task-b", "tenant-a", "tenant-key-a", 1, "running"), - )); err != nil { - t.Fatal(err) - } -} - -func localSnapshotWindow(dispatcherID, cursor string, body []byte, tasks ...LocalDiscoveredTask) LocalTaskDiscovery { - return LocalTaskDiscovery{DispatcherID: dispatcherID, Mode: "snapshot", Cursor: cursor, Body: body, Tasks: tasks} -} - -func localDiscoveredTask(_ string, taskID, tenantID, tenantKey string, revision int64, status string) LocalDiscoveredTask { - return LocalDiscoveredTask{TaskID: taskID, TenantID: tenantID, TenantKey: tenantKey, TaskRevision: revision, Status: status} -} diff --git a/internal/store/local_v03_admission_test.go b/internal/store/local_v03_admission_test.go new file mode 100644 index 0000000..c667826 --- /dev/null +++ b/internal/store/local_v03_admission_test.go @@ -0,0 +1,40 @@ +package store + +import ( + "errors" + "testing" + "time" +) + +func TestEventDiscoveryAdmissionRequiresRunningTask(t *testing.T) { + for _, tc := range []struct { + name string + discovered bool + status string + wantErr error + }{ + {name: "paused", discovered: true, status: "paused", wantErr: ErrLocalTaskPaused}, + {name: "stopped", discovered: true, status: "stopped", wantErr: ErrLocalTaskStopped}, + {name: "removed", discovered: true, status: "removed", wantErr: ErrLocalTaskStopped}, + {name: "no discovery", wantErr: ErrLocalDiscoveryUnavailable}, + } { + t.Run(tc.name, func(t *testing.T) { + st := openLocalDiscoveryTestStore(t) + if tc.discovered { + seedLocalEventTasks(t, st, "d-1", eventTask("task-a", tc.status, 1)) + } + now := time.Date(2026, 9, 21, 2, 0, 0, 0, time.UTC) + bundle := localConfigBundle("d-1", "tenant-a", "tenant-key-a", "task-a", 1, 1, 1, 1, 1, now) + record := localAdmissionCommand("command-a", "execution-a", "task-a", bundle, "d-1", "tenant-a", "tenant-key-a") + if _, err := st.PersistLocalCommand(record); !errors.Is(err, tc.wantErr) { + t.Fatalf("PersistLocalCommand error=%v, want %v", err, tc.wantErr) + } + for _, table := range []string{"inbox", "outbox", "tasks", "reservations"} { + var count int + if err := st.db.QueryRow(`SELECT COUNT(*) FROM ` + table).Scan(&count); err != nil || count != 0 { + t.Fatalf("table %s rows=%d err=%v after blocked admission", table, count, err) + } + } + }) + } +} diff --git a/internal/store/local_v03_config_gate_test.go b/internal/store/local_v03_config_gate_test.go new file mode 100644 index 0000000..d3d81d0 --- /dev/null +++ b/internal/store/local_v03_config_gate_test.go @@ -0,0 +1,35 @@ +package store + +import ( + "errors" + "testing" +) + +func TestEventConfigReadRequiresCompleteDiscoveryAndTenantBinding(t *testing.T) { + st := openLocalDiscoveryTestStore(t) + if err := st.ApplyLocalTaskDiscoveryPage(eventPage("0", "1", eventTask("task-a", "running", 1))); err != nil { + t.Fatal(err) + } + if _, err := st.LocalDiscoveredTaskForConfig("d-1", "task-a", "tenant-a"); !errors.Is(err, ErrLocalDiscoveryUnavailable) { + t.Fatalf("partial discovery admitted config read: %v", err) + } + if err := st.ApplyLocalTaskDiscoveryPage(eventPage("1", "1")); err != nil { + t.Fatal(err) + } + assignment, err := st.LocalDiscoveredTaskForConfig("d-1", "task-a", "tenant-a") + if err != nil || assignment.Status != "running" || assignment.TenantKey != "tenant-key-a" { + t.Fatalf("assignment=%+v err=%v", assignment, err) + } + if _, err := st.LocalDiscoveredTaskForConfig("d-1", "task-a", "another-tenant"); !errors.Is(err, ErrTenantBindingConflict) { + t.Fatalf("cross-tenant config read: %v", err) + } + if err := st.ApplyLocalTaskDiscoveryPage(eventPage("1", "2", eventTask("task-a", "removed", 1))); err != nil { + t.Fatal(err) + } + if err := st.ApplyLocalTaskDiscoveryPage(eventPage("2", "2")); err != nil { + t.Fatal(err) + } + if _, err := st.LocalDiscoveredTaskForConfig("d-1", "task-a", "tenant-a"); !errors.Is(err, ErrLocalTaskRemoved) { + t.Fatalf("removed task config read: %v", err) + } +} diff --git a/internal/store/local_v03_discovery.go b/internal/store/local_v03_discovery.go new file mode 100644 index 0000000..7651189 --- /dev/null +++ b/internal/store/local_v03_discovery.go @@ -0,0 +1,165 @@ +package store + +import ( + "crypto/sha256" + "database/sql" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "strconv" + "time" +) + +const localEventPageLimit = 256 + +// LocalTaskDiscoveryPage is one complete HTTP response, not an entire task list. +// SaaS guarantees response ordering/completeness; its response-level cursor is +// the only event position supplied by the project-local v0.3 proposal. +type LocalTaskDiscoveryPage struct { + DispatcherID string + FromCursor string + NextCursor string + Tasks []LocalDiscoveredTask + Body json.RawMessage + ObservedAt time.Time +} + +func eventCursorValue(cursor string) (uint64, error) { + value, err := strconv.ParseUint(cursor, 10, 64) + if err != nil || strconv.FormatUint(value, 10) != cursor { + return 0, errors.New("event cursor must be a canonical decimal uint64") + } + return value, nil +} + +// ApplyLocalTaskDiscoveryPage commits assignments, the exact page evidence and +// its cursor together. A nonempty page never opens new execution admission; +// only an empty final page marks the initial catch-up/current poll complete. +func (s *Store) ApplyLocalTaskDiscoveryPage(page LocalTaskDiscoveryPage) error { + from, err := eventCursorValue(page.FromCursor) + if err != nil { + return fmt.Errorf("invalid task-discovery after cursor: %w", err) + } + to, err := eventCursorValue(page.NextCursor) + if err != nil { + return fmt.Errorf("invalid task-discovery next cursor: %w", err) + } + if page.DispatcherID == "" || page.ObservedAt.IsZero() || len(page.Tasks) > localEventPageLimit || !json.Valid(page.Body) || + (len(page.Tasks) == 0 && to != from) || (len(page.Tasks) != 0 && to <= from) { + return ErrLocalDiscoveryUnavailable + } + seen := make(map[string]struct{}, len(page.Tasks)) + for _, task := range page.Tasks { + if err := validateLocalDiscoveredTask(page.DispatcherID, task); err != nil { + return err + } + if _, duplicate := seen[task.TaskID]; duplicate { + return fmt.Errorf("task-discovery page has duplicate task ID %q", task.TaskID) + } + seen[task.TaskID] = struct{}{} + } + + s.mu.Lock() + defer s.mu.Unlock() + tx, err := s.db.Begin() + if err != nil { + return err + } + defer tx.Rollback() + var legacy int + if err := tx.QueryRow(`SELECT COUNT(*) FROM local_v02_task_discovery_state WHERE dispatcher_id=?`, page.DispatcherID).Scan(&legacy); err != nil { + return err + } + if legacy != 0 { + return fmt.Errorf("%w: v0.2 discovery checkpoint must be drained before switching", ErrLocalDiscoveryUnavailable) + } + var cursor string + err = tx.QueryRow(`SELECT cursor FROM local_v03_task_discovery_state WHERE dispatcher_id=?`, page.DispatcherID).Scan(&cursor) + if errors.Is(err, sql.ErrNoRows) { + var existing int + if err := tx.QueryRow(`SELECT COUNT(*) FROM local_v01_task_assignments WHERE dispatcher_id=?`, page.DispatcherID).Scan(&existing); err != nil { + return err + } + if existing != 0 { + return fmt.Errorf("%w: existing assignments have no v0.3 checkpoint", ErrLocalDiscoveryUnavailable) + } + if page.FromCursor != "0" { + return ErrLocalDiscoveryCursorMismatch + } + } else if err != nil { + return err + } else if cursor != page.FromCursor { + return ErrLocalDiscoveryCursorMismatch + } + now := page.ObservedAt.UTC().Format(time.RFC3339Nano) + for _, task := range page.Tasks { + if err := bindLocalTenant(tx, page.DispatcherID, task.TenantID, task.TenantKey, page.ObservedAt); err != nil { + return err + } + if err := applyLocalDiscoveredTask(tx, page.DispatcherID, task, now); err != nil { + return err + } + } + var active int + if err := tx.QueryRow(`SELECT COUNT(*) FROM local_v01_task_assignments WHERE dispatcher_id=? AND removed=0`, page.DispatcherID).Scan(&active); err != nil { + return err + } + if active > localEventPageLimit { + return ErrLocalDiscoveryTaskLimit + } + ready := 0 + if len(page.Tasks) == 0 { + ready = 1 + } + sum := sha256.Sum256(page.Body) + _, err = tx.Exec(`INSERT INTO local_v03_task_discovery_state(dispatcher_id,cursor,ready,body,content_sha256,updated_at) + VALUES(?,?,?,?,?,?) ON CONFLICT(dispatcher_id) DO UPDATE SET cursor=excluded.cursor,ready=excluded.ready, + body=excluded.body,content_sha256=excluded.content_sha256,updated_at=excluded.updated_at`, + page.DispatcherID, page.NextCursor, ready, []byte(page.Body), hex.EncodeToString(sum[:]), now) + if err != nil { + return err + } + return tx.Commit() +} + +// LocalDiscoveredTaskForConfig binds a config read to a fully caught-up, +// durable task assignment. A partial page cannot authorize configuration. +func (s *Store) LocalDiscoveredTaskForConfig(dispatcherID, taskID, tenantID string) (LocalTaskAssignment, error) { + s.mu.Lock() + defer s.mu.Unlock() + var ready int + var observedAt string + err := s.db.QueryRow(`SELECT ready,updated_at FROM local_v03_task_discovery_state WHERE dispatcher_id=?`, dispatcherID).Scan(&ready, &observedAt) + if errors.Is(err, sql.ErrNoRows) || (err == nil && ready != 1) { + return LocalTaskAssignment{}, ErrLocalDiscoveryUnavailable + } + if err != nil { + return LocalTaskAssignment{}, err + } + var assignment LocalTaskAssignment + var removed int + err = s.db.QueryRow(`SELECT dispatcher_id,task_id,tenant_id,tenant_key,task_revision,saas_status,admission_state,removed, + queue_exchange,routing_key,binding_key,queue_name,updated_at FROM local_v01_task_assignments WHERE dispatcher_id=? AND task_id=?`, + dispatcherID, taskID).Scan(&assignment.DispatcherID, &assignment.TaskID, &assignment.TenantID, &assignment.TenantKey, + &assignment.TaskRevision, &assignment.Status, &assignment.AdmissionState, &removed, &assignment.Queue.Exchange, + &assignment.Queue.RoutingKey, &assignment.Queue.BindingKey, &assignment.Queue.QueueName, &assignment.UpdatedAt) + if errors.Is(err, sql.ErrNoRows) { + return LocalTaskAssignment{}, ErrLocalDiscoveryTaskMissing + } + if err != nil { + return LocalTaskAssignment{}, err + } + if assignment.TenantID != tenantID { + return LocalTaskAssignment{}, ErrTenantBindingConflict + } + assignment.Removed = removed != 0 + if assignment.Removed { + return LocalTaskAssignment{}, ErrLocalTaskRemoved + } + assignment.DiscoveryObservedAt, err = time.Parse(time.RFC3339Nano, observedAt) + if err != nil { + return LocalTaskAssignment{}, fmt.Errorf("invalid durable discovery observation time: %w", err) + } + return assignment, nil +} diff --git a/internal/store/local_v03_discovery_test.go b/internal/store/local_v03_discovery_test.go new file mode 100644 index 0000000..01b219f --- /dev/null +++ b/internal/store/local_v03_discovery_test.go @@ -0,0 +1,92 @@ +package store + +import ( + "crypto/sha256" + "encoding/hex" + "errors" + "testing" + "time" +) + +func eventPage(after, next string, tasks ...LocalDiscoveredTask) LocalTaskDiscoveryPage { + return LocalTaskDiscoveryPage{ + DispatcherID: "d-1", FromCursor: after, NextCursor: next, + Tasks: tasks, Body: []byte(`{"schema_version":"task-discovery.v0.3-proposal","tasks":[]}`), + ObservedAt: time.Now().UTC(), + } +} + +func eventTask(id, status string, revision int64) LocalDiscoveredTask { + return LocalDiscoveredTask{TaskID: id, TenantID: "tenant-a", TenantKey: "tenant-key-a", Status: status, TaskRevision: revision} +} + +func TestEventPagesPersistCursorAssignmentAndBootGateTogether(t *testing.T) { + st := openLocalDiscoveryTestStore(t) + first := eventPage("0", "1", eventTask("task-a", "running", 1)) + if err := st.ApplyLocalTaskDiscoveryPage(first); err != nil { + t.Fatal(err) + } + cursor, exists, err := st.LocalTaskDiscoveryCursor("d-1") + if err != nil || !exists || cursor != "1" { + t.Fatalf("cursor=%q exists=%v err=%v", cursor, exists, err) + } + var ready int + var saved []byte + var digest string + if err := st.db.QueryRow(`SELECT ready,body,content_sha256 FROM local_v03_task_discovery_state WHERE dispatcher_id=?`, "d-1").Scan(&ready, &saved, &digest); err != nil { + t.Fatal(err) + } + sum := sha256.Sum256(first.Body) + if ready != 0 || string(saved) != string(first.Body) || digest != hex.EncodeToString(sum[:]) { + t.Fatalf("partial page admitted or changed evidence: ready=%d body=%s hash=%s", ready, saved, digest) + } + assignments, err := st.LocalTaskAssignments("d-1") + if err != nil || len(assignments) != 1 || assignments[0].Status != "running" || assignments[0].Queue.QueueName != "agent-call.d.d-1.task.task-a.v3" { + t.Fatalf("assignment=%+v err=%v", assignments, err) + } + if err := st.ApplyLocalTaskDiscoveryPage(eventPage("1", "1")); err != nil { + t.Fatal(err) + } + if err := st.db.QueryRow(`SELECT ready FROM local_v03_task_discovery_state WHERE dispatcher_id=?`, "d-1").Scan(&ready); err != nil || ready != 1 { + t.Fatalf("completed page ready=%d err=%v", ready, err) + } +} + +func TestEventPageAllowsSameTaskNewStatusAndRemovedTombstone(t *testing.T) { + st := openLocalDiscoveryTestStore(t) + for _, page := range []LocalTaskDiscoveryPage{ + eventPage("0", "1", eventTask("task-a", "running", 1)), + eventPage("1", "2", eventTask("task-a", "paused", 1)), + eventPage("2", "3", eventTask("task-a", "removed", 1)), + eventPage("3", "3"), + } { + if err := st.ApplyLocalTaskDiscoveryPage(page); err != nil { + t.Fatal(err) + } + } + assignments, err := st.LocalTaskAssignments("d-1") + if err != nil || len(assignments) != 1 || !assignments[0].Removed || assignments[0].Status != "removed" || assignments[0].AdmissionState != "removed" { + t.Fatalf("removed task=%+v err=%v", assignments, err) + } + cursor, exists, err := st.LocalTaskDiscoveryCursor("d-1") + if err != nil || !exists || cursor != "3" { + t.Fatalf("cursor=%q exists=%v err=%v", cursor, exists, err) + } + if err := st.ApplyLocalTaskDiscoveryPage(eventPage("1", "4", eventTask("task-a", "running", 1))); !errors.Is(err, ErrLocalDiscoveryCursorMismatch) { + t.Fatalf("old cursor unexpectedly accepted: %v", err) + } +} + +func TestEventPageCannotInterpretOldSnapshotCursor(t *testing.T) { + st := openLocalDiscoveryTestStore(t) + if _, err := st.db.Exec(`INSERT INTO local_v02_task_discovery_state(dispatcher_id,cursor,last_mode,ready,updated_at) VALUES('d-1','opaque-old','snapshot',1,'now')`); err != nil { + t.Fatal(err) + } + if err := st.ApplyLocalTaskDiscoveryPage(eventPage("0", "1", eventTask("task-a", "running", 1))); !errors.Is(err, ErrLocalDiscoveryUnavailable) { + t.Fatalf("legacy cursor should block, got %v", err) + } + var count int + if err := st.db.QueryRow(`SELECT COUNT(*) FROM local_v03_task_discovery_state WHERE dispatcher_id='d-1'`).Scan(&count); err != nil || count != 0 { + t.Fatalf("legacy cursor created new checkpoint: count=%d err=%v", count, err) + } +} diff --git a/internal/store/local_v03_failure_test.go b/internal/store/local_v03_failure_test.go new file mode 100644 index 0000000..fd5faa9 --- /dev/null +++ b/internal/store/local_v03_failure_test.go @@ -0,0 +1,78 @@ +package store + +import ( + "errors" + "strings" + "testing" +) + +func TestEventPageWriteFailureRollsBackAssignmentCursorAndEvidence(t *testing.T) { + st := openLocalDiscoveryTestStore(t) + if err := st.ApplyLocalTaskDiscoveryPage(eventPage("0", "1", eventTask("task-a", "running", 1))); err != nil { + t.Fatal(err) + } + var before []byte + if err := st.db.QueryRow(`SELECT body FROM local_v03_task_discovery_state WHERE dispatcher_id='d-1'`).Scan(&before); err != nil { + t.Fatal(err) + } + if _, err := st.db.Exec(`CREATE TRIGGER reject_discovery_update BEFORE UPDATE ON local_v01_task_assignments + BEGIN SELECT RAISE(ABORT, 'forced assignment write failure'); END`); err != nil { + t.Fatal(err) + } + err := st.ApplyLocalTaskDiscoveryPage(eventPage("1", "2", eventTask("task-a", "paused", 2))) + if err == nil || !strings.Contains(err.Error(), "forced assignment write failure") { + t.Fatalf("assignment write failure was swallowed: %v", err) + } + cursor, exists, err := st.LocalTaskDiscoveryCursor("d-1") + if err != nil || !exists || cursor != "1" { + t.Fatalf("failed page advanced cursor: %q exists=%v err=%v", cursor, exists, err) + } + assignment, err := st.LocalTaskAssignment("d-1", "task-a") + if err != nil || assignment.Status != "running" || assignment.TaskRevision != 1 { + t.Fatalf("failed page changed assignment: %+v err=%v", assignment, err) + } + var after []byte + if err := st.db.QueryRow(`SELECT body FROM local_v03_task_discovery_state WHERE dispatcher_id='d-1'`).Scan(&after); err != nil || string(after) != string(before) { + t.Fatalf("failed page replaced saved evidence: before=%s after=%s err=%v", before, after, err) + } +} + +func TestEventPageCannotUndoMQPauseStopOrRemoval(t *testing.T) { + st := openLocalDiscoveryTestStore(t) + if err := st.ApplyLocalTaskDiscoveryPage(eventPage("0", "1", eventTask("task-a", "running", 1))); err != nil { + t.Fatal(err) + } + if _, err := st.SetLocalTaskAdmissionBarrier("d-1", "task-a", "tenant-a", "tenant-key-a", "paused"); err != nil { + t.Fatal(err) + } + if err := st.ApplyLocalTaskDiscoveryPage(eventPage("1", "2", eventTask("task-a", "running", 2))); err != nil { + t.Fatal(err) + } + assignment, err := st.LocalTaskAssignment("d-1", "task-a") + if err != nil || assignment.AdmissionState != "paused" { + t.Fatalf("stale running discovery overrode MQ pause: %+v err=%v", assignment, err) + } + if _, err := st.SetLocalTaskAdmissionBarrier("d-1", "task-a", "tenant-a", "tenant-key-a", "stopped"); err != nil { + t.Fatal(err) + } + if err := st.ApplyLocalTaskDiscoveryPage(eventPage("2", "3", eventTask("task-a", "running", 3))); err != nil { + t.Fatal(err) + } + assignment, err = st.LocalTaskAssignment("d-1", "task-a") + if err != nil || assignment.AdmissionState != "stopped" { + t.Fatalf("stale running discovery overrode MQ stop: %+v err=%v", assignment, err) + } + if err := st.ApplyLocalTaskDiscoveryPage(eventPage("3", "4", eventTask("task-a", "removed", 3))); err != nil { + t.Fatal(err) + } + if err := st.ApplyLocalTaskDiscoveryPage(eventPage("4", "5", eventTask("task-a", "running", 4))); !errors.Is(err, ErrLocalTaskRemoved) { + t.Fatalf("removed task was revived: %v", err) + } + if err := st.ApplyLocalTaskDiscoveryPage(eventPage("4", "4")); err != nil { + t.Fatal(err) + } + assignments, err := st.LocalTaskAssignments("d-1") + if err != nil || len(assignments) != 1 || !assignments[0].Removed || assignments[0].AdmissionState != "removed" { + t.Fatalf("later event revived removed task: %+v err=%v", assignments, err) + } +} diff --git a/internal/store/migrations/018_local_v03_task_discovery.sql b/internal/store/migrations/018_local_v03_task_discovery.sql new file mode 100644 index 0000000..240ecc6 --- /dev/null +++ b/internal/store/migrations/018_local_v03_task_discovery.sql @@ -0,0 +1,10 @@ +-- Separate event-cursor checkpoint: never reinterpret a v0.2 snapshot/change cursor. +-- The older tables remain untouched for controlled drain/diagnosis, not fallback. +CREATE TABLE IF NOT EXISTS local_v03_task_discovery_state ( + dispatcher_id TEXT PRIMARY KEY, + cursor TEXT NOT NULL, + ready INTEGER NOT NULL CHECK (ready IN (0, 1)), + body BLOB NOT NULL, + content_sha256 TEXT NOT NULL, + updated_at TEXT NOT NULL +);