Remove plugin data query projection

This commit is contained in:
npc0-hue
2026-08-21 15:36:49 +08:00
parent fc86b24699
commit 8aeb488179
20 changed files with 91 additions and 703 deletions
-6
View File
@@ -40,9 +40,6 @@ func (svc *CoreService) ClaimRunJob(claim domain.RunJobClaim) (domain.RunJobClai
if err := svc.sweepExpiredJobs(claim.RunEndpointID, stamp); err != nil {
return domain.RunJobClaimResult{}, err
}
if err := svc.scheduleDuePluginQueries(claim.RunEndpointID, claim.Capabilities, stamp); err != nil {
return domain.RunJobClaimResult{}, err
}
if claim.Capacity.MaxJobs > 0 && claim.Capacity.RunningJobs >= claim.Capacity.MaxJobs {
return emptyJobClaim(claim.RunEndpointID, stamp), nil
}
@@ -263,9 +260,6 @@ func (svc *CoreService) CompleteRunJob(result domain.RunJobResult) (domain.RunJo
if err := svc.validateDistributionBuildResult(job); err != nil {
return domain.RunJobResultResult{}, err
}
if err := svc.projectPluginDataJobResult(job); err != nil {
return domain.RunJobResultResult{}, err
}
if err := svc.updateScheduledJob(job); err != nil {
return domain.RunJobResultResult{}, err
}
-126
View File
@@ -1,126 +0,0 @@
package service
import (
"encoding/json"
"errors"
"fmt"
"sort"
"strings"
"browser.local/platform/domain"
"browser.local/platform/repo"
)
type pluginDataQueryResult struct {
Rows []map[string]any `json:"rows"`
}
func (svc *CoreService) projectPluginDataJobResult(job domain.Job) error {
if job.Capability != domain.JobCapabilityRemoteRunDBSQLiteQuery || job.State != domain.JobStateSucceeded {
return nil
}
templateKey := strings.TrimSpace(job.ExecutionInput.Inputs["templateKey"])
if templateKey == "" {
return nil
}
instance, err := svc.store.ServerInstances().Get(job.ServerInstanceID)
if err != nil {
return err
}
plugin, err := svc.store.GamePlugins().Get(instance.PluginID)
if err != nil {
return err
}
var template domain.GameClientBridgeQueryTemplateDeclaration
for _, candidate := range plugin.GameClientBridge.QueryTemplates {
if candidate.Key == templateKey {
template = candidate
break
}
}
if template.RowTarget == nil {
return nil
}
var result pluginDataQueryResult
if err := json.Unmarshal([]byte(job.ExecutionResult.Content), &result); err != nil {
return validationError("declared query result is not valid JSON")
}
mutationsByKey := make(map[string]domain.PluginDataMutation, len(result.Rows))
for _, row := range result.Rows {
value := make(map[string]any, len(template.RowTarget.ColumnMappings))
for destination, source := range template.RowTarget.ColumnMappings {
value[destination] = row[source]
}
key, err := pluginDataRowKey(value, template.RowTarget.UpsertKeys)
if err != nil {
return err
}
if template.RowTarget.WriteMode == domain.PluginDataRowWriteModeMerge {
existing, getErr := svc.store.PluginDataRecords().Get(pluginDataID(instance.ID, plugin.ID, template.RowTarget.Collection, key))
if getErr == nil {
value = mergePluginDataValues(existing.Value, value)
} else if !errors.Is(getErr, repo.ErrNotFound) {
return getErr
}
}
mutationsByKey[key] = domain.PluginDataMutation{Operation: domain.PluginDataMutationPut, Key: key, Value: value}
}
if template.RowTarget.WriteMode == domain.PluginDataRowWriteModeReplace {
existing, listErr := svc.store.PluginDataRecords().List(domain.PluginDataFilter{PluginID: plugin.ID, ServerInstanceID: instance.ID, Collection: template.RowTarget.Collection})
if listErr != nil {
return listErr
}
for _, record := range existing {
if _, present := mutationsByKey[record.Key]; !present {
mutationsByKey[record.Key] = domain.PluginDataMutation{Operation: domain.PluginDataMutationDelete, Key: record.Key}
}
}
}
if len(mutationsByKey) == 0 {
return nil
}
keys := make([]string, 0, len(mutationsByKey))
for key := range mutationsByKey {
keys = append(keys, key)
}
sort.Strings(keys)
mutations := make([]domain.PluginDataMutation, 0, len(keys))
for _, key := range keys {
mutations = append(mutations, mutationsByKey[key])
}
_, err = svc.applyPluginDataTransaction(domain.PluginDataTransaction{PluginID: plugin.ID, ServerInstanceID: instance.ID, Collection: template.RowTarget.Collection, Mutations: mutations})
return err
}
func mergePluginDataValues(existing, incoming map[string]any) map[string]any {
merged := domain.CopyGameClientBridgePayload(existing)
if merged == nil {
merged = make(map[string]any, len(incoming))
}
for key, value := range incoming {
merged[key] = value
}
return merged
}
func pluginDataRowKey(value map[string]any, keys []string) (string, error) {
parts := make([]string, len(keys))
for index, key := range keys {
item, exists := value[key]
if !exists || item == nil || strings.TrimSpace(fmt.Sprint(item)) == "" {
return "", validationError("declared query row is missing an upsert key")
}
encoded, err := json.Marshal(item)
if err != nil {
return "", validationError("declared query row upsert key is invalid")
}
parts[index] = string(encoded)
}
if len(parts) == 1 {
var key string
if err := json.Unmarshal([]byte(parts[0]), &key); err == nil {
return key, nil
}
}
return strings.Join(parts, "\x1f"), nil
}
+42
View File
@@ -0,0 +1,42 @@
package service
import (
"encoding/json"
"fmt"
"strings"
"browser.local/platform/domain"
)
func mergePluginDataValues(existing, incoming map[string]any) map[string]any {
merged := domain.CopyGameClientBridgePayload(existing)
if merged == nil {
merged = make(map[string]any, len(incoming))
}
for key, value := range incoming {
merged[key] = value
}
return merged
}
func pluginDataRowKey(value map[string]any, keys []string) (string, error) {
parts := make([]string, len(keys))
for index, key := range keys {
item, exists := value[key]
if !exists || item == nil || strings.TrimSpace(fmt.Sprint(item)) == "" {
return "", validationError("declared plugin data row is missing an upsert key")
}
encoded, err := json.Marshal(item)
if err != nil {
return "", validationError("declared plugin data row upsert key is invalid")
}
parts[index] = string(encoded)
}
if len(parts) == 1 {
var key string
if err := json.Unmarshal([]byte(parts[0]), &key); err == nil {
return key, nil
}
}
return strings.Join(parts, "\x1f"), nil
}
-92
View File
@@ -1,92 +0,0 @@
package service
import (
"fmt"
"strconv"
"strings"
"time"
"browser.local/platform/domain"
)
func (svc *CoreService) scheduleDuePluginQueries(runEndpointID string, capabilities []string, stamp time.Time) error {
if !containsString(capabilities, domain.JobCapabilityRemoteRunDBSQLiteQuery) {
return nil
}
instances, err := svc.store.ServerInstances().List(domain.ServerInstanceFilter{RunEndpointID: runEndpointID})
if err != nil {
return err
}
jobs, err := svc.store.Jobs().List(domain.JobFilter{RunEndpointID: runEndpointID})
if err != nil {
return err
}
for _, instance := range instances {
if instance.State == domain.ServerInstanceStateDeleted {
continue
}
plugin, getErr := svc.store.GamePlugins().Get(instance.PluginID)
if getErr != nil {
return getErr
}
for _, template := range plugin.GameClientBridge.QueryTemplates {
if template.PollIntervalSeconds <= 0 || template.RowTarget == nil || strings.TrimSpace(template.SQLRef) == "" {
continue
}
if !pluginQueryTemplateDue(jobs, instance.ID, template.Key, time.Duration(template.PollIntervalSeconds)*time.Second, stamp) {
continue
}
bucket := stamp.Unix() / int64(template.PollIntervalSeconds)
idempotencyKey := fmt.Sprintf("plugin-query:%s:%s:%d", instance.ID, template.Key, bucket)
job := domain.Job{
ID: jobIDFromParts("job-plugin-query", instance.ID, idempotencyKey),
ServerInstanceID: instance.ID,
RunEndpointID: runEndpointID,
Capability: domain.JobCapabilityRemoteRunDBSQLiteQuery,
TargetKey: template.TargetKey,
InputRef: "input://plugin-query/" + template.Key,
IdempotencyKey: idempotencyKey,
Progress: domain.JobProgress{Percent: 0, Message: "declared automatic plugin query queued"},
RetryPolicy: domain.JobRetryPolicy{MaxAttempts: 1, InitialBackoffSeconds: 1, MaxBackoffSeconds: 1},
ExecutionInput: domain.JobExecutionInput{
WorkspaceScope: svc.runtimeProfileScope(instance.ID),
RemoteAdapterKey: template.TransportKey,
RemoteAdapterKind: string(domain.RemoteAdapterDatabase),
TimeoutSeconds: template.TimeoutSeconds,
Inputs: map[string]string{
"templateKey": template.Key,
"sqlRef": template.SQLRef,
"maxRows": strconv.Itoa(template.MaxRows),
"limit": strconv.Itoa(template.MaxRows),
},
},
}
created, createErr := svc.CreateJob(job)
if createErr != nil {
return createErr
}
jobs = append(jobs, created)
}
}
return nil
}
func pluginQueryTemplateDue(jobs []domain.Job, serverInstanceID, templateKey string, interval time.Duration, stamp time.Time) bool {
var latest time.Time
for _, job := range jobs {
if job.ServerInstanceID != serverInstanceID || job.Capability != domain.JobCapabilityRemoteRunDBSQLiteQuery || job.ExecutionInput.Inputs["templateKey"] != templateKey {
continue
}
if !isTerminalJobState(job.State) {
return false
}
attemptedAt := job.TerminalAt
if attemptedAt.IsZero() {
attemptedAt = job.UpdatedAt
}
if attemptedAt.After(latest) {
latest = attemptedAt
}
}
return latest.IsZero() || !stamp.Before(latest.Add(interval))
}
+21 -164
View File
@@ -98,28 +98,20 @@ func TestPluginDataTransactionValidationFailureDoesNotPartiallyApply(t *testing.
}
}
func TestDeclaredSQLiteQueryProjectsRowsIntoPluginCollection(t *testing.T) {
func TestDeclaredSQLiteQueryDoesNotMutatePluginOrPlatformUserData(t *testing.T) {
svc, plugin, endpoint, session, instance := createSQLiteQueryBridgeFixture(t)
plugin.GameClientBridge.QueryTemplates[0].RowTarget = &domain.PluginDataRowTargetDeclaration{
Collection: "users",
UpsertKeys: []string{"userId"},
ColumnMappings: map[string]string{
"userId": "user_id",
"displayName": "display_name",
},
if _, err := svc.PutPluginDataForSession(session, domain.PluginDataRecord{PluginID: plugin.ID, ServerInstanceID: instance.ID, Collection: "scum_users", Key: "steam-keep", Value: map[string]any{"steamId": "steam-keep", "format": "plugin-local"}}); err != nil {
t.Fatalf("seed plugin data: %v", err)
}
if err := svc.store.GamePlugins().Update(plugin); err != nil {
t.Fatalf("update plugin row target: %v", err)
}
queued, err := svc.ExecutePluginBridgeAction(session, domain.PluginBridgeExecuteRequest{RequestID: "query-project-1", PluginID: plugin.ID, RouteKey: "remote", ServerInstanceID: instance.ID, Action: domain.PluginBridgeActionRemoteAccessRequest, Payload: map[string]string{
"capability": domain.JobCapabilityRemoteRunDBSQLiteQuery, "declarationKey": "scum-db-read", "targetKey": "scum-db.player-lookup", "idempotencyKey": "query-project-1", "input.templateKey": "players.by-id",
queued, err := svc.ExecutePluginBridgeAction(session, domain.PluginBridgeExecuteRequest{RequestID: "query-no-projection-1", PluginID: plugin.ID, RouteKey: "remote", ServerInstanceID: instance.ID, Action: domain.PluginBridgeActionRemoteAccessRequest, Payload: map[string]string{
"capability": domain.JobCapabilityRemoteRunDBSQLiteQuery, "declarationKey": "scum-db-read", "targetKey": "scum-db.player-lookup", "idempotencyKey": "query-no-projection-1", "input.templateKey": "players.by-id",
}})
if err != nil || queued.Status != "queued" {
t.Fatalf("queue declared query=%+v err=%v", queued, err)
}
helloRequest := validRunControlHello()
helloRequest.CapabilityReport.Capabilities = append(helloRequest.CapabilityReport.Capabilities, domain.JobCapabilityRemoteRunDBSQLiteQuery)
helloRequest.CapabilityReport.Fingerprint = "cap-plugin-data-query"
helloRequest.CapabilityReport.Fingerprint = "cap-plugin-data-query-independent"
hello, err := svc.RegisterRunHello(helloRequest)
if err != nil {
t.Fatalf("register Run: %v", err)
@@ -132,170 +124,35 @@ func TestDeclaredSQLiteQueryProjectsRowsIntoPluginCollection(t *testing.T) {
if err != nil {
t.Fatalf("complete query job: %v", err)
}
items, err := svc.ListPluginDataForSession(session, domain.PluginDataFilter{PluginID: plugin.ID, ServerInstanceID: instance.ID, Collection: "users"})
if err != nil || len(items) != 2 || items[0].Key != "steam-1" || items[0].Value["displayName"] != "Ada" {
t.Fatalf("projected plugin rows=%+v err=%v", items, err)
items, err := svc.ListPluginDataForSession(session, domain.PluginDataFilter{PluginID: plugin.ID, ServerInstanceID: instance.ID, Collection: "scum_users"})
if err != nil || len(items) != 1 || items[0].Key != "steam-keep" || items[0].Value["format"] != "plugin-local" {
t.Fatalf("declared query mutated plugin data=%+v err=%v", items, err)
}
current, err := svc.GetCurrentUser(session)
if err != nil || current.ID != "user-query-owner" || current.Email != "query-owner@example.test" {
t.Fatalf("declared query mutated platform user session/user=%+v err=%v", current, err)
}
}
func TestDeclaredSQLiteQueryProjectionRejectsInvalidBatchAtomically(t *testing.T) {
svc, plugin, _, session, instance := createSQLiteQueryBridgeFixture(t)
plugin.GameClientBridge.QueryTemplates[0].RowTarget = &domain.PluginDataRowTargetDeclaration{
Collection: "members",
UpsertKeys: []string{"accountId"},
ColumnMappings: map[string]string{
"accountId": "account_id",
"displayName": "display_name",
},
}
if err := svc.store.GamePlugins().Update(plugin); err != nil {
t.Fatalf("update plugin row target: %v", err)
}
job := domain.Job{
ServerInstanceID: instance.ID,
Capability: domain.JobCapabilityRemoteRunDBSQLiteQuery,
State: domain.JobStateSucceeded,
ExecutionInput: domain.JobExecutionInput{Inputs: map[string]string{"templateKey": plugin.GameClientBridge.QueryTemplates[0].Key}},
ExecutionResult: domain.JobExecutionResult{Content: `{"rows":[{"account_id":"one","display_name":"Ada","ignored":"value"},{"display_name":"Missing key"}]}`},
}
if err := svc.projectPluginDataJobResult(job); err == nil {
t.Fatal("expected missing upsert key error")
}
items, err := svc.ListPluginDataForSession(session, domain.PluginDataFilter{PluginID: plugin.ID, ServerInstanceID: instance.ID, Collection: "members"})
if err != nil || len(items) != 0 {
t.Fatalf("invalid projection batch partially applied values=%+v err=%v", items, err)
}
}
func TestDeclaredSQLiteQueryProjectionMergesPresenceAndReplacesCompleteSnapshots(t *testing.T) {
svc, plugin, _, session, instance := createSQLiteQueryBridgeFixture(t)
template := &plugin.GameClientBridge.QueryTemplates[0]
template.RowTarget = &domain.PluginDataRowTargetDeclaration{
Collection: "users", UpsertKeys: []string{"steamId"}, WriteMode: domain.PluginDataRowWriteModeMerge,
ColumnMappings: map[string]string{"steamId": "steam_id", "displayName": "display_name", "x": "x"},
}
if err := svc.store.GamePlugins().Update(plugin); err != nil {
t.Fatalf("update merge target: %v", err)
}
if _, err := svc.PutPluginDataForSession(session, domain.PluginDataRecord{PluginID: plugin.ID, ServerInstanceID: instance.ID, Collection: "users", Key: "steam-1", Value: map[string]any{"steamId": "steam-1", "online": true, "lastLoginAt": "2026-07-03T12:00:00Z"}}); err != nil {
t.Fatalf("seed stdout user: %v", err)
}
job := domain.Job{ServerInstanceID: instance.ID, Capability: domain.JobCapabilityRemoteRunDBSQLiteQuery, State: domain.JobStateSucceeded, ExecutionInput: domain.JobExecutionInput{Inputs: map[string]string{"templateKey": template.Key}}, ExecutionResult: domain.JobExecutionResult{Content: `{"rows":[{"steam_id":"steam-1","display_name":"Ada","x":12.5}]}`}}
if err := svc.projectPluginDataJobResult(job); err != nil {
t.Fatalf("merge query projection: %v", err)
}
users, err := svc.ListPluginDataForSession(session, domain.PluginDataFilter{PluginID: plugin.ID, ServerInstanceID: instance.ID, Collection: "users"})
if err != nil || len(users) != 1 || users[0].Value["online"] != true || users[0].Value["displayName"] != "Ada" {
t.Fatalf("merged users=%+v err=%v", users, err)
}
plugin.GameClientBridge.QueryTemplates[0].RowTarget = &domain.PluginDataRowTargetDeclaration{
Collection: "vehicles", UpsertKeys: []string{"vehicleId"}, WriteMode: domain.PluginDataRowWriteModeReplace,
ColumnMappings: map[string]string{"vehicleId": "vehicle_id", "x": "x"},
}
if err := svc.store.GamePlugins().Update(plugin); err != nil {
t.Fatalf("update replace target: %v", err)
}
for _, id := range []string{"keep", "gone"} {
if _, err := svc.PutPluginDataForSession(session, domain.PluginDataRecord{PluginID: plugin.ID, ServerInstanceID: instance.ID, Collection: "vehicles", Key: id, Value: map[string]any{"vehicleId": id}}); err != nil {
t.Fatalf("seed vehicle %s: %v", id, err)
}
}
job.ExecutionResult.Content = `{"rows":[{"vehicle_id":"keep","x":7}]}`
if err := svc.projectPluginDataJobResult(job); err != nil {
t.Fatalf("replace query projection: %v", err)
}
vehicles, err := svc.ListPluginDataForSession(session, domain.PluginDataFilter{PluginID: plugin.ID, ServerInstanceID: instance.ID, Collection: "vehicles"})
if err != nil || len(vehicles) != 1 || vehicles[0].Key != "keep" {
t.Fatalf("replaced vehicles=%+v err=%v", vehicles, err)
}
job.ExecutionResult.Content = `{"rows":[]}`
if err := svc.projectPluginDataJobResult(job); err != nil {
t.Fatalf("empty replace query projection: %v", err)
}
vehicles, err = svc.ListPluginDataForSession(session, domain.PluginDataFilter{PluginID: plugin.ID, ServerInstanceID: instance.ID, Collection: "vehicles"})
if err != nil || len(vehicles) != 0 {
t.Fatalf("empty replace did not clear vehicles=%+v err=%v", vehicles, err)
}
}
func TestRunPollSchedulesDueDeclaredPluginQueryWithoutBrowserSession(t *testing.T) {
svc, plugin, endpoint, _, instance := createSQLiteQueryBridgeFixture(t)
func TestRunPollDoesNotSchedulePluginDataProjectionQueries(t *testing.T) {
svc, plugin, endpoint, _, _ := createSQLiteQueryBridgeFixture(t)
plugin.GameClientBridge.QueryTemplates[0].PollIntervalSeconds = 3
if err := svc.store.GamePlugins().Update(plugin); err != nil {
t.Fatalf("enable automatic query: %v", err)
t.Fatalf("enable query refresh hint: %v", err)
}
helloRequest := validRunControlHello()
helloRequest.CapabilityReport.Capabilities = append(helloRequest.CapabilityReport.Capabilities, domain.JobCapabilityRemoteRunDBSQLiteQuery)
helloRequest.CapabilityReport.Fingerprint = "cap-plugin-query-scheduler"
helloRequest.CapabilityReport.Fingerprint = "cap-plugin-query-no-scheduler"
hello, err := svc.RegisterRunHello(helloRequest)
if err != nil {
t.Fatalf("register Run: %v", err)
}
claim, err := svc.ClaimRunJob(domain.RunJobClaim{RunEndpointID: endpoint.ID, SessionToken: hello.SessionToken, Capabilities: []string{domain.JobCapabilityRemoteRunDBSQLiteQuery}, Capacity: domain.RunCapacity{MaxJobs: 1}})
if err != nil || !claim.HasJob || claim.Job == nil {
t.Fatalf("automatic query claim=%+v err=%v", claim, err)
}
if claim.Job.ServerInstanceID != instance.ID || claim.Job.Capability != domain.JobCapabilityRemoteRunDBSQLiteQuery || claim.Job.ExecutionInput.Inputs["templateKey"] != "players.by-id" || claim.Job.ExecutionInput.Inputs["limit"] != "25" {
t.Fatalf("unexpected automatic query assignment: %+v", claim.Job)
if err != nil || claim.HasJob {
t.Fatalf("automatic projection query was scheduled: %+v err=%v", claim, err)
}
jobs, err := svc.store.Jobs().List(domain.JobFilter{RunEndpointID: endpoint.ID})
if err != nil || len(jobs) != 1 {
t.Fatalf("automatic query jobs=%+v err=%v", jobs, err)
}
second, err := svc.ClaimRunJob(domain.RunJobClaim{RunEndpointID: endpoint.ID, SessionToken: hello.SessionToken, Capabilities: []string{domain.JobCapabilityRemoteRunDBSQLiteQuery}, Capacity: domain.RunCapacity{MaxJobs: 1, RunningJobs: 1}})
if err != nil || second.HasJob {
t.Fatalf("overlapping automatic query was not suppressed: %+v err=%v", second, err)
}
jobs, err = svc.store.Jobs().List(domain.JobFilter{RunEndpointID: endpoint.ID})
if err != nil || len(jobs) != 1 {
t.Fatalf("overlap created duplicate jobs=%+v err=%v", jobs, err)
}
}
func TestDeclaredSQLiteQueryProjectionFailureKeepsJobRetryable(t *testing.T) {
svc, plugin, endpoint, session, instance := createSQLiteQueryBridgeFixture(t)
plugin.GameClientBridge.QueryTemplates[0].RowTarget = &domain.PluginDataRowTargetDeclaration{
Collection: "members",
UpsertKeys: []string{"accountId"},
ColumnMappings: map[string]string{
"accountId": "account_id",
"displayName": "display_name",
},
}
if err := svc.store.GamePlugins().Update(plugin); err != nil {
t.Fatalf("update plugin row target: %v", err)
}
queued, err := svc.ExecutePluginBridgeAction(session, domain.PluginBridgeExecuteRequest{RequestID: "query-invalid-projection", PluginID: plugin.ID, RouteKey: "remote", ServerInstanceID: instance.ID, Action: domain.PluginBridgeActionRemoteAccessRequest, Payload: map[string]string{
"capability": domain.JobCapabilityRemoteRunDBSQLiteQuery, "declarationKey": "scum-db-read", "targetKey": "scum-db.player-lookup", "idempotencyKey": "query-invalid-projection", "input.templateKey": "players.by-id",
}})
if err != nil || queued.Status != "queued" {
t.Fatalf("queue declared query=%+v err=%v", queued, err)
}
helloRequest := validRunControlHello()
helloRequest.CapabilityReport.Capabilities = append(helloRequest.CapabilityReport.Capabilities, domain.JobCapabilityRemoteRunDBSQLiteQuery)
helloRequest.CapabilityReport.Fingerprint = "cap-plugin-data-invalid-query"
hello, err := svc.RegisterRunHello(helloRequest)
if err != nil {
t.Fatalf("register Run: %v", err)
}
claim, err := svc.ClaimRunJob(domain.RunJobClaim{RunEndpointID: endpoint.ID, SessionToken: hello.SessionToken, Capabilities: []string{domain.JobCapabilityRemoteRunDBSQLiteQuery}, Capacity: domain.RunCapacity{MaxJobs: 1}})
if err != nil || !claim.HasJob || claim.Job == nil {
t.Fatalf("claim query job=%+v err=%v", claim, err)
}
result := domain.RunJobResult{RunEndpointID: endpoint.ID, SessionToken: hello.SessionToken, JobID: claim.Job.JobID, LeaseToken: claim.Job.LeaseToken, Attempt: claim.Job.Attempt, State: domain.JobStateSucceeded, Progress: domain.RunJobProgressReport{Percent: 100}, ExecutionResult: domain.JobExecutionResult{Kind: "sqlite.query", Content: `{"rows":[{"display_name":"Missing key"}]}`}}
if _, err := svc.CompleteRunJob(result); err == nil {
t.Fatal("expected projection failure")
}
job, err := svc.store.Jobs().Get(claim.Job.JobID)
if err != nil || isTerminalJobState(job.State) {
t.Fatalf("projection failure persisted terminal job=%+v err=%v", job, err)
}
if _, err := svc.CompleteRunJob(result); err == nil {
t.Fatal("expected projection retry to re-run and fail")
}
items, err := svc.ListPluginDataForSession(session, domain.PluginDataFilter{PluginID: plugin.ID, ServerInstanceID: instance.ID, Collection: "members"})
if err != nil || len(items) != 0 {
t.Fatalf("invalid retry projected records=%+v err=%v", items, err)
if err != nil || len(jobs) != 0 {
t.Fatalf("automatic projection query persisted jobs=%+v err=%v", jobs, err)
}
}
-1
View File
@@ -1846,7 +1846,6 @@ func createSQLiteQueryBridgeFixture(t *testing.T) (*CoreService, domain.GamePlug
SQLRef: "sql/players.by-id.sql",
MaxRows: 25,
TimeoutSeconds: 20,
RowTarget: &domain.PluginDataRowTargetDeclaration{Collection: "players", UpsertKeys: []string{"userId"}, ColumnMappings: map[string]string{"userId": "user_id"}},
},
},
Retention: domain.GameClientBridgeRetention{KeepForSeconds: 3600, MaxRecords: 100},