diff --git a/platform/README.md b/platform/README.md index e0f155b..300cb87 100644 --- a/platform/README.md +++ b/platform/README.md @@ -47,7 +47,7 @@ Runtime configuration: - `PLATFORM_STORAGE_BACKEND`: storage backend, default `file`; use `memory` only for tests or disposable local runs. - `PLATFORM_MYSQL_DSN`: MySQL DSN used when `PLATFORM_STORAGE_BACKEND=mysql`, for example `platform:platform@tcp(127.0.0.1:3306)/platform?parseTime=true`. - `PLATFORM_DATA_DIR`: default platform data directory, default `.platform-data`. -- `PLATFORM_METADATA_PATH`: file-backed metadata snapshot path, default `.platform-data/metadata.json`. +- `PLATFORM_METADATA_PATH`: file-backed metadata snapshot path, default `.platform-data/metadata.json`; high-frequency Run endpoint/session state uses a sibling runtime snapshot to avoid rewriting the full metadata file on every heartbeat. - `PLATFORM_LOG_BODY_BACKEND`: log body backend, default follows metadata backend except MySQL uses `file`; supported values are `file` and `memory`. - `PLATFORM_LOG_DIR`: segmented log body directory, default `.platform-data/logs`. - `PLATFORM_ARTIFACT_DIR`: private durable artifact body/transfer directory, default `.platform-data/artifacts`. diff --git a/platform/api/resource_handlers.go b/platform/api/resource_handlers.go index fc6402d..bdff957 100644 --- a/platform/api/resource_handlers.go +++ b/platform/api/resource_handlers.go @@ -2437,14 +2437,19 @@ func (h *coreHandlers) runEndpointDetail(w http.ResponseWriter, r *http.Request) func (h *coreHandlers) jobs(w http.ResponseWriter, r *http.Request) { switch r.Method { case http.MethodGet: + limit, err := optionalPositiveInt(r.URL.Query().Get("limit")) + if err != nil || limit < 0 { + writeAPIError(w, http.StatusBadRequest, errorCodeBadRequest, "invalid job limit", nil) + return + } filter := domain.JobFilter{ ServerInstanceID: r.URL.Query().Get("serverInstanceId"), RunEndpointID: r.URL.Query().Get("runEndpointId"), State: domain.JobState(r.URL.Query().Get("state")), States: parseJobStates(r.URL.Query().Get("states")), + Limit: limit, } var jobs []domain.Job - var err error if h.enforceAuthorization { jobs, err = h.core.ListJobsForSession(bearerToken(r), filter) } else { diff --git a/platform/api/resource_handlers_test.go b/platform/api/resource_handlers_test.go index e1d8df1..8c4020a 100644 --- a/platform/api/resource_handlers_test.go +++ b/platform/api/resource_handlers_test.go @@ -88,6 +88,8 @@ func TestCoreAPICreateListDetailWorkflows(t *testing.T) { assertListCount(t, jobs.Count, 1) jobs = getJSON[dto.JobListResponse](t, router, "/api/v1/jobs?serverInstanceId=server-1&states=queued,running,failed") assertListCount(t, jobs.Count, 1) + jobs = getJSON[dto.JobListResponse](t, router, "/api/v1/jobs?serverInstanceId=server-1&states=queued,running,failed&limit=1") + assertListCount(t, jobs.Count, 1) jobs = getJSON[dto.JobListResponse](t, router, "/api/v1/jobs?serverInstanceId=server-1&states=running,failed") assertListCount(t, jobs.Count, 0) diff --git a/platform/api/routes.md b/platform/api/routes.md index 1a08e04..b8167e6 100644 --- a/platform/api/routes.md +++ b/platform/api/routes.md @@ -38,7 +38,7 @@ Plugin-owned data is an independent, server-scoped plugin store. It is not a pro - `GET /api/v1/metrics/server-instances` - `GET /api/v1/run/endpoints?status=online` - `GET /api/v1/jobs?serverInstanceId=server-1&runEndpointId=run-local&state=queued` -- `GET /api/v1/jobs?serverInstanceId=server-1&states=queued,running,failed` +- `GET /api/v1/jobs?serverInstanceId=server-1&states=queued,running,failed&limit=100` - `GET /api/v1/artifacts?ownerKind=job&ownerId=job-1&state=uploading` - `GET /api/v1/log-streams?serverInstanceId=server-1&streamKey=stdout` @@ -181,7 +181,7 @@ Control is the highest-priority run-facing channel; artifact/file transfer press - `POST /api/v1/run/jobs/reconcile`: accept persisted Run journal evidence (`jobId`, `attempt`, `leaseToken`), rebind only matching active attempts to the current authenticated session generation, persist reconciliation metadata, retry/cancel platform-active missing work, and return confirmed assignments plus discard IDs. - `POST /api/v1/jobs/{id}/cancel`: authorize the server owner/administrator or platform administrator and durably record cancellation intent; queued/retrying work becomes cancelled immediately while active work completes through fenced Run polling/result. -Run job actions carry bounded job metadata only: job ID, run endpoint ID, server instance ID, capability, idempotency key, lease token, attempt/retry limits, deadlines, progress, terminal state, message, error code, result reference, and timing hints. Raw lease tokens exist only on the signed Run job channel; platform persistence stores their hashes. User-facing Job responses expose safe attempt, retry, cancel, terminal, and reconcile projections but never raw/hashed leases, Run sessions, secret refs, host paths, sockets, or credentials. +Run job actions carry bounded job metadata only: job ID, run endpoint ID, server instance ID, capability, idempotency key, lease token, attempt/retry limits, deadlines, progress, terminal state, message, error code, result reference, and timing hints. Raw lease tokens exist only on the signed Run job channel; platform persistence stores their hashes. User-facing Job detail responses expose safe execution-result evidence; Job list responses stay summary-only for hot console polling. Job responses never expose raw/hashed leases, Run sessions, secret refs, host paths, sockets, or credentials. Job ack/progress/result/cancel/reconcile calls remain lightweight and independently valid while log batches or artifact/file chunks are queued, slow, or retrying. Equivalent duplicate terminal results remain idempotent under channel pressure. ## Implemented Run File Transfer Actions diff --git a/platform/domain/resources.go b/platform/domain/resources.go index dec5d25..20442a3 100644 --- a/platform/domain/resources.go +++ b/platform/domain/resources.go @@ -1597,6 +1597,7 @@ type JobFilter struct { RunEndpointID string State JobState States []JobState + Limit int } type ArtifactFilter struct { diff --git a/platform/dto/resources.go b/platform/dto/resources.go index 7366a61..9cd4777 100644 --- a/platform/dto/resources.go +++ b/platform/dto/resources.go @@ -888,31 +888,31 @@ type JobRetryPolicyResponse struct { } type JobResponse struct { - ID string `json:"id"` - ServerInstanceID string `json:"serverInstanceId,omitempty"` - RunEndpointID string `json:"runEndpointId"` - Capability string `json:"capability"` - TargetKey string `json:"targetKey,omitempty"` - InputRef string `json:"inputRef,omitempty"` - IdempotencyKey string `json:"idempotencyKey"` - State domain.JobState `json:"state"` - Progress JobProgressBody `json:"progress"` - ResultRef string `json:"resultRef,omitempty"` - ExecutionResult JobExecutionResultResponse `json:"executionResult,omitempty"` - RetryPolicy JobRetryPolicyResponse `json:"retryPolicy"` - Attempt int `json:"attempt"` - NextAttemptAt *time.Time `json:"nextAttemptAt,omitempty"` - AckDeadlineAt *time.Time `json:"ackDeadlineAt,omitempty"` - LeaseExpiresAt *time.Time `json:"leaseExpiresAt,omitempty"` - CancelReason string `json:"cancelReason,omitempty"` - CancelRequestedAt *time.Time `json:"cancelRequestedAt,omitempty"` - CancelCompletedAt *time.Time `json:"cancelCompletedAt,omitempty"` - TerminalAt *time.Time `json:"terminalAt,omitempty"` - LastReconciledAt *time.Time `json:"lastReconciledAt,omitempty"` - ReconcileCount int `json:"reconcileCount"` - ReconcileOutcome string `json:"reconcileOutcome,omitempty"` - CreatedAt time.Time `json:"createdAt"` - UpdatedAt time.Time `json:"updatedAt"` + ID string `json:"id"` + ServerInstanceID string `json:"serverInstanceId,omitempty"` + RunEndpointID string `json:"runEndpointId"` + Capability string `json:"capability"` + TargetKey string `json:"targetKey,omitempty"` + InputRef string `json:"inputRef,omitempty"` + IdempotencyKey string `json:"idempotencyKey"` + State domain.JobState `json:"state"` + Progress JobProgressBody `json:"progress"` + ResultRef string `json:"resultRef,omitempty"` + ExecutionResult *JobExecutionResultResponse `json:"executionResult,omitempty"` + RetryPolicy JobRetryPolicyResponse `json:"retryPolicy"` + Attempt int `json:"attempt"` + NextAttemptAt *time.Time `json:"nextAttemptAt,omitempty"` + AckDeadlineAt *time.Time `json:"ackDeadlineAt,omitempty"` + LeaseExpiresAt *time.Time `json:"leaseExpiresAt,omitempty"` + CancelReason string `json:"cancelReason,omitempty"` + CancelRequestedAt *time.Time `json:"cancelRequestedAt,omitempty"` + CancelCompletedAt *time.Time `json:"cancelCompletedAt,omitempty"` + TerminalAt *time.Time `json:"terminalAt,omitempty"` + LastReconciledAt *time.Time `json:"lastReconciledAt,omitempty"` + ReconcileCount int `json:"reconcileCount"` + ReconcileOutcome string `json:"reconcileOutcome,omitempty"` + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` } type JobExecutionResultResponse struct { @@ -1990,6 +1990,18 @@ func RunEndpointListFromDomain(endpoints []domain.RunEndpoint) RunEndpointListRe } func JobFromDomain(job domain.Job) JobResponse { + return jobResponseFromDomain(job, true) +} + +func JobListItemFromDomain(job domain.Job) JobResponse { + return jobResponseFromDomain(job, false) +} + +func jobResponseFromDomain(job domain.Job, includeExecutionResult bool) JobResponse { + var executionResult *JobExecutionResultResponse + if includeExecutionResult { + executionResult = jobExecutionResultFromDomain(job.ExecutionResult) + } return JobResponse{ ID: job.ID, ServerInstanceID: job.ServerInstanceID, @@ -2001,7 +2013,7 @@ func JobFromDomain(job domain.Job) JobResponse { State: job.State, Progress: progressFromDomain(job.Progress), ResultRef: job.ResultRef, - ExecutionResult: JobExecutionResultResponse{Kind: job.ExecutionResult.Kind, ProcessState: job.ExecutionResult.ProcessState, ExitClassification: job.ExecutionResult.ExitClassification, ExitCode: job.ExecutionResult.ExitCode, Version: job.ExecutionResult.Version, Checksum: job.ExecutionResult.Checksum, SizeBytes: job.ExecutionResult.SizeBytes, Summary: job.ExecutionResult.Summary, ServerDeploymentEvidence: serverDeploymentEvidenceFromDomain(job.ExecutionResult.ServerDeploymentEvidence)}, + ExecutionResult: executionResult, RetryPolicy: JobRetryPolicyResponse{ MaxAttempts: job.RetryPolicy.MaxAttempts, InitialBackoffSeconds: job.RetryPolicy.InitialBackoffSeconds, @@ -2023,6 +2035,14 @@ func JobFromDomain(job domain.Job) JobResponse { } } +func jobExecutionResultFromDomain(result domain.JobExecutionResult) *JobExecutionResultResponse { + response := JobExecutionResultResponse{Kind: result.Kind, ProcessState: result.ProcessState, ExitClassification: result.ExitClassification, ExitCode: result.ExitCode, Version: result.Version, Checksum: result.Checksum, SizeBytes: result.SizeBytes, Summary: result.Summary, ServerDeploymentEvidence: serverDeploymentEvidenceFromDomain(result.ServerDeploymentEvidence)} + if response.Kind == "" && response.ProcessState == "" && response.ExitClassification == "" && response.ExitCode == 0 && response.Version == 0 && response.Checksum == "" && response.SizeBytes == 0 && response.Summary == "" && response.ServerDeploymentEvidence == nil { + return nil + } + return &response +} + func optionalTime(value time.Time) *time.Time { if value.IsZero() { return nil @@ -2034,7 +2054,7 @@ func optionalTime(value time.Time) *time.Time { func JobListFromDomain(jobs []domain.Job) JobListResponse { items := make([]JobResponse, len(jobs)) for i, job := range jobs { - items[i] = JobFromDomain(job) + items[i] = JobListItemFromDomain(job) } return JobListResponse{Items: items, Count: len(items)} } diff --git a/platform/protocol/run-contracts.md b/platform/protocol/run-contracts.md index 9937e17..3944edf 100644 --- a/platform/protocol/run-contracts.md +++ b/platform/protocol/run-contracts.md @@ -89,7 +89,7 @@ Run-assigned Platform jobs use `job..` log stream IDs. Autonom Log ingest is durable and independently retried. Artifact/file transfer backlog must not prevent log batch acknowledgement, duplicate acknowledgement, cursor state updates, or spool cleanup. -The platform stores log stream metadata through `repo.Store` and stores log bodies through the configured `LogBodyStore`. The default `file` backend persists platform metadata to `PLATFORM_METADATA_PATH` and appends log entries to segmented JSONL files under `PLATFORM_LOG_DIR`; the `memory` backend is only for tests and disposable local development. MySQL/Postgres are appropriate for platform metadata, stream state, retention policy, indexes, and operational records, but should not be the primary row-per-log-line store for hundreds or thousands of servers. Production log bodies should move behind the same boundary to append/query backends such as ClickHouse, Loki, OpenSearch/Elasticsearch, or object-storage segments with compact indexes. +The platform stores log stream metadata through `repo.Store` and stores log bodies through the configured `LogBodyStore`. The default `file` backend persists platform metadata to `PLATFORM_METADATA_PATH`, keeps high-frequency Run endpoint/session runtime state in a lightweight sibling runtime snapshot, and appends log entries to segmented JSONL files under `PLATFORM_LOG_DIR`; the `memory` backend is only for tests and disposable local development. MySQL/Postgres are appropriate for platform metadata, stream state, retention policy, indexes, and operational records, but should not be the primary row-per-log-line store for hundreds or thousands of servers. Production log bodies should move behind the same boundary to append/query backends such as ClickHouse, Loki, OpenSearch/Elasticsearch, or object-storage segments with compact indexes. ## Artifact diff --git a/platform/repo/file_store.go b/platform/repo/file_store.go index ca798b3..f5a4d30 100644 --- a/platform/repo/file_store.go +++ b/platform/repo/file_store.go @@ -38,10 +38,17 @@ type StoreSnapshot struct { PluginDataRecords []domain.PluginDataRecord `json:"pluginDataRecords"` } +type runtimeSnapshot struct { + RunControlSessions []domain.RunControlSession `json:"runControlSessions"` + RunEndpoints []domain.RunEndpoint `json:"runEndpoints"` +} + type FileStore struct { *MemoryStore - path string - persistMu sync.Mutex + path string + runtimePath string + persistMu sync.Mutex + runtimePersistMu sync.Mutex } func NewFileStore(path string) (*FileStore, error) { @@ -52,6 +59,7 @@ func NewFileStore(path string) (*FileStore, error) { store := &FileStore{ MemoryStore: NewMemoryStore(), path: path, + runtimePath: runtimeMetadataPath(path), } if err := store.load(); err != nil { return nil, err @@ -72,7 +80,7 @@ func (store *FileStore) AuthSessions() AuthSessionRepository { } func (store *FileStore) RunControlSessions() RunControlSessionRepository { - return &persistentRepository[domain.RunControlSession, struct{}]{repository: store.MemoryStore.runSessions, persist: store.persist} + return &persistentRepository[domain.RunControlSession, struct{}]{repository: store.MemoryStore.runSessions, persist: store.persistRuntime} } func (store *FileStore) AIProviders() AIProviderRepository { @@ -88,7 +96,7 @@ func (store *FileStore) ServerInstances() ServerInstanceRepository { } func (store *FileStore) RunEndpoints() RunEndpointRepository { - return &persistentRepository[domain.RunEndpoint, domain.RunEndpointFilter]{repository: store.MemoryStore.runEndpoints, persist: store.persist} + return &persistentRepository[domain.RunEndpoint, domain.RunEndpointFilter]{repository: store.MemoryStore.runEndpoints, persist: store.persistRuntime} } func (store *FileStore) Jobs() JobRepository { @@ -163,19 +171,18 @@ func (store *FileStore) load() error { data, err := os.ReadFile(store.path) if err != nil { if os.IsNotExist(err) { - return nil + return store.loadRuntime() } return fmt.Errorf("read metadata snapshot: %w", err) } - if len(strings.TrimSpace(string(data))) == 0 { - return nil + if len(strings.TrimSpace(string(data))) != 0 { + var snapshot StoreSnapshot + if err := json.Unmarshal(data, &snapshot); err != nil { + return fmt.Errorf("decode metadata snapshot: %w", err) + } + store.loadSnapshot(snapshot) } - var snapshot StoreSnapshot - if err := json.Unmarshal(data, &snapshot); err != nil { - return fmt.Errorf("decode metadata snapshot: %w", err) - } - store.loadSnapshot(snapshot) - return nil + return store.loadRuntime() } func (store *FileStore) persist() error { @@ -183,7 +190,7 @@ func (store *FileStore) persist() error { defer store.persistMu.Unlock() snapshot := store.snapshot() - data, err := json.MarshalIndent(snapshot, "", " ") + data, err := json.Marshal(snapshot) if err != nil { return fmt.Errorf("encode metadata snapshot: %w", err) } @@ -200,6 +207,85 @@ func (store *FileStore) persist() error { return nil } +func (store *FileStore) loadRuntime() error { + data, err := os.ReadFile(store.runtimePath) + if err != nil { + if os.IsNotExist(err) { + return nil + } + return fmt.Errorf("read runtime metadata snapshot: %w", err) + } + if len(strings.TrimSpace(string(data))) == 0 { + return nil + } + var snapshot runtimeSnapshot + if err := json.Unmarshal(data, &snapshot); err != nil { + return fmt.Errorf("decode runtime metadata snapshot: %w", err) + } + store.loadRuntimeSnapshot(snapshot) + return nil +} + +func (store *FileStore) persistRuntime() error { + store.runtimePersistMu.Lock() + defer store.runtimePersistMu.Unlock() + + snapshot := store.runtimeSnapshot() + data, err := json.Marshal(snapshot) + if err != nil { + return fmt.Errorf("encode runtime metadata snapshot: %w", err) + } + if err := os.MkdirAll(filepath.Dir(store.runtimePath), 0o755); err != nil { + return fmt.Errorf("create runtime metadata directory: %w", err) + } + if err := store.ensureMetadataFile(); err != nil { + return err + } + tmpPath := store.runtimePath + ".tmp" + if err := os.WriteFile(tmpPath, data, 0o600); err != nil { + return fmt.Errorf("write runtime metadata snapshot: %w", err) + } + if err := os.Rename(tmpPath, store.runtimePath); err != nil { + return fmt.Errorf("replace runtime metadata snapshot: %w", err) + } + return nil +} + +func (store *FileStore) ensureMetadataFile() error { + if _, err := os.Stat(store.path); err == nil { + return nil + } else if !os.IsNotExist(err) { + return fmt.Errorf("stat metadata snapshot: %w", err) + } + + store.persistMu.Lock() + defer store.persistMu.Unlock() + if _, err := os.Stat(store.path); err == nil { + return nil + } else if !os.IsNotExist(err) { + return fmt.Errorf("stat metadata snapshot: %w", err) + } + if err := os.MkdirAll(filepath.Dir(store.path), 0o755); err != nil { + return fmt.Errorf("create metadata directory: %w", err) + } + tmpPath := store.path + ".empty.tmp" + if err := os.WriteFile(tmpPath, []byte("{}"), 0o600); err != nil { + return fmt.Errorf("write empty metadata snapshot: %w", err) + } + if err := os.Rename(tmpPath, store.path); err != nil { + return fmt.Errorf("replace empty metadata snapshot: %w", err) + } + return nil +} + +func runtimeMetadataPath(path string) string { + ext := filepath.Ext(path) + if ext == "" { + return path + ".runtime" + } + return strings.TrimSuffix(path, ext) + ".runtime" + ext +} + func (store *FileStore) snapshot() StoreSnapshot { return StoreSnapshot{ Users: snapshotRepository(store.MemoryStore.users), @@ -228,6 +314,13 @@ func (store *FileStore) snapshot() StoreSnapshot { } } +func (store *FileStore) runtimeSnapshot() runtimeSnapshot { + return runtimeSnapshot{ + RunControlSessions: snapshotRepository(store.MemoryStore.runSessions), + RunEndpoints: snapshotRepository(store.MemoryStore.runEndpoints), + } +} + func (store *FileStore) loadSnapshot(snapshot StoreSnapshot) { loadRepository(store.MemoryStore.users, snapshot.Users) loadRepository(store.MemoryStore.authSessions, snapshot.AuthSessions) @@ -254,6 +347,11 @@ func (store *FileStore) loadSnapshot(snapshot StoreSnapshot) { loadRepository(store.MemoryStore.pluginDataRecords, snapshot.PluginDataRecords) } +func (store *FileStore) loadRuntimeSnapshot(snapshot runtimeSnapshot) { + loadRepository(store.MemoryStore.runSessions, snapshot.RunControlSessions) + loadRepository(store.MemoryStore.runEndpoints, snapshot.RunEndpoints) +} + type mutableRepository[T any, F any] interface { Create(T) error Get(string) (T, error) diff --git a/platform/repo/mysql_store.go b/platform/repo/mysql_store.go index 91251b0..2f4d8a3 100644 --- a/platform/repo/mysql_store.go +++ b/platform/repo/mysql_store.go @@ -14,12 +14,16 @@ import ( _ "github.com/go-sql-driver/mysql" ) -const mysqlSnapshotID = "current" +const ( + mysqlSnapshotID = "current" + mysqlRuntimeSnapshotID = "runtime" +) type MySQLStore struct { *MemoryStore - db *sql.DB - persistMu sync.Mutex + db *sql.DB + persistMu sync.Mutex + runtimePersistMu sync.Mutex } func NewMySQLStore(dsn string) (*MySQLStore, error) { @@ -59,7 +63,7 @@ func (store *MySQLStore) AuthSessions() AuthSessionRepository { } func (store *MySQLStore) RunControlSessions() RunControlSessionRepository { - return &persistentRepository[domain.RunControlSession, struct{}]{repository: store.MemoryStore.runSessions, persist: store.persist} + return &persistentRepository[domain.RunControlSession, struct{}]{repository: store.MemoryStore.runSessions, persist: store.persistRuntime} } func (store *MySQLStore) AIProviders() AIProviderRepository { @@ -75,7 +79,7 @@ func (store *MySQLStore) ServerInstances() ServerInstanceRepository { } func (store *MySQLStore) RunEndpoints() RunEndpointRepository { - return &persistentRepository[domain.RunEndpoint, domain.RunEndpointFilter]{repository: store.MemoryStore.runEndpoints, persist: store.persist} + return &persistentRepository[domain.RunEndpoint, domain.RunEndpointFilter]{repository: store.MemoryStore.runEndpoints, persist: store.persistRuntime} } func (store *MySQLStore) Jobs() JobRepository { @@ -171,7 +175,7 @@ func (store *MySQLStore) load() error { err := store.db.QueryRowContext(ctx, "SELECT snapshot_json FROM platform_metadata_snapshots WHERE id = ?", mysqlSnapshotID).Scan(&payload) if err != nil { if err == sql.ErrNoRows { - return nil + return store.loadRuntime() } return fmt.Errorf("read mysql metadata snapshot: %w", err) } @@ -180,6 +184,25 @@ func (store *MySQLStore) load() error { return fmt.Errorf("decode mysql metadata snapshot: %w", err) } store.loadSnapshot(snapshot) + return store.loadRuntime() +} + +func (store *MySQLStore) loadRuntime() error { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + var payload []byte + err := store.db.QueryRowContext(ctx, "SELECT snapshot_json FROM platform_metadata_snapshots WHERE id = ?", mysqlRuntimeSnapshotID).Scan(&payload) + if err != nil { + if err == sql.ErrNoRows { + return nil + } + return fmt.Errorf("read mysql runtime metadata snapshot: %w", err) + } + var snapshot runtimeSnapshot + if err := json.Unmarshal(payload, &snapshot); err != nil { + return fmt.Errorf("decode mysql runtime metadata snapshot: %w", err) + } + store.loadRuntimeSnapshot(snapshot) return nil } @@ -204,6 +227,27 @@ ON DUPLICATE KEY UPDATE snapshot_json = ?, updated_at = CURRENT_TIMESTAMP`, mysq return nil } +func (store *MySQLStore) persistRuntime() error { + store.runtimePersistMu.Lock() + defer store.runtimePersistMu.Unlock() + + snapshot := store.runtimeSnapshot() + payload, err := json.Marshal(snapshot) + if err != nil { + return fmt.Errorf("encode mysql runtime metadata snapshot: %w", err) + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _, err = store.db.ExecContext(ctx, ` +INSERT INTO platform_metadata_snapshots (id, snapshot_json) +VALUES (?, ?) +ON DUPLICATE KEY UPDATE snapshot_json = ?, updated_at = CURRENT_TIMESTAMP`, mysqlRuntimeSnapshotID, string(payload), string(payload)) + if err != nil { + return fmt.Errorf("write mysql runtime metadata snapshot: %w", err) + } + return nil +} + func (store *MySQLStore) snapshot() StoreSnapshot { return StoreSnapshot{ Users: snapshotRepository(store.MemoryStore.users), @@ -232,6 +276,13 @@ func (store *MySQLStore) snapshot() StoreSnapshot { } } +func (store *MySQLStore) runtimeSnapshot() runtimeSnapshot { + return runtimeSnapshot{ + RunControlSessions: snapshotRepository(store.MemoryStore.runSessions), + RunEndpoints: snapshotRepository(store.MemoryStore.runEndpoints), + } +} + func (store *MySQLStore) loadSnapshot(snapshot StoreSnapshot) { loadRepository(store.MemoryStore.users, snapshot.Users) loadRepository(store.MemoryStore.authSessions, snapshot.AuthSessions) @@ -257,3 +308,8 @@ func (store *MySQLStore) loadSnapshot(snapshot StoreSnapshot) { loadRepository(store.MemoryStore.bridgeStreams, snapshot.GameClientBridgeStreams) loadRepository(store.MemoryStore.pluginDataRecords, snapshot.PluginDataRecords) } + +func (store *MySQLStore) loadRuntimeSnapshot(snapshot runtimeSnapshot) { + loadRepository(store.MemoryStore.runSessions, snapshot.RunControlSessions) + loadRepository(store.MemoryStore.runEndpoints, snapshot.RunEndpoints) +} diff --git a/platform/repo/resources.go b/platform/repo/resources.go index 777163d..5c3b872 100644 --- a/platform/repo/resources.go +++ b/platform/repo/resources.go @@ -486,6 +486,39 @@ func newMemoryJobRepository() *memoryJobRepository { } } +func (repository *memoryJobRepository) List(filter domain.JobFilter) ([]domain.Job, error) { + if filter.Limit <= 0 { + return repository.memoryRepository.List(filter) + } + + repository.mu.RLock() + defer repository.mu.RUnlock() + + values := make([]domain.Job, 0, min(filter.Limit, len(repository.byID))) + for _, job := range repository.byID { + if repository.match(job, filter) { + values = append(values, domain.CopyJob(job)) + } + } + sort.SliceStable(values, func(i, j int) bool { + leftUpdated := values[i].UpdatedAt + rightUpdated := values[j].UpdatedAt + if !leftUpdated.Equal(rightUpdated) { + return leftUpdated.After(rightUpdated) + } + leftCreated := values[i].CreatedAt + rightCreated := values[j].CreatedAt + if !leftCreated.Equal(rightCreated) { + return leftCreated.After(rightCreated) + } + return values[i].ID < values[j].ID + }) + if len(values) > filter.Limit { + values = values[:filter.Limit] + } + return values, nil +} + func (repository *memoryJobRepository) GetByIdempotency(runEndpointID string, idempotencyKey string) (domain.Job, error) { repository.mu.RLock() defer repository.mu.RUnlock() diff --git a/platform/repo/resources_test.go b/platform/repo/resources_test.go index bc245b4..d2eb43f 100644 --- a/platform/repo/resources_test.go +++ b/platform/repo/resources_test.go @@ -1,6 +1,7 @@ package repo import ( + "bytes" "encoding/json" "errors" "os" @@ -306,6 +307,56 @@ func TestMySQLSnapshotRoundTripsDurableJobSchedulingMetadata(t *testing.T) { } } +func TestFileStorePersistsRunRuntimeStateSeparately(t *testing.T) { + path := filepath.Join(t.TempDir(), "metadata.json") + store, err := NewFileStore(path) + if err != nil { + t.Fatalf("create file store: %v", err) + } + stamp := time.Date(2026, 7, 18, 10, 0, 0, 0, time.UTC) + if err := store.PluginDataRecords().Create(domain.PluginDataRecord{ID: "plugin-record-1", PluginID: "server.scum", ServerInstanceID: "server-1", Collection: "players", Key: "steam-1", Value: map[string]any{"displayName": "Ada"}, CreatedAt: stamp, UpdatedAt: stamp}); err != nil { + t.Fatalf("create plugin data: %v", err) + } + mainBefore, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read main snapshot before runtime update: %v", err) + } + + endpoint := domain.RunEndpoint{ID: "run-runtime", DisplayName: "Runtime Run", Version: "1.0.0", Platform: "linux", Architecture: "amd64", Status: domain.RunEndpointStatusOnline, Capacity: domain.RunCapacity{MaxJobs: 4}, LastHeartbeatAt: stamp} + if err := store.RunEndpoints().Create(endpoint); err != nil { + t.Fatalf("create runtime endpoint: %v", err) + } + session := domain.RunControlSession{RunEndpointID: endpoint.ID, SessionToken: "raw-runtime-token", SessionTokenHash: strings.Repeat("a", 64), Status: domain.AuthSessionStatusActive, Generation: 1, CapabilityFingerprint: "cap-runtime", HeartbeatIntervalSeconds: 15, CreatedAt: stamp, UpdatedAt: stamp, ExpiresAt: stamp.Add(time.Hour), RequireSignedRequests: true, UsedNonces: []string{"nonce-1"}} + if err := store.RunControlSessions().Create(session); err != nil { + t.Fatalf("create runtime session: %v", err) + } + mainAfter, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read main snapshot after runtime update: %v", err) + } + if !bytes.Equal(mainBefore, mainAfter) { + t.Fatalf("runtime updates rewrote the main metadata snapshot") + } + runtimePayload, err := os.ReadFile(runtimeMetadataPath(path)) + if err != nil { + t.Fatalf("read runtime snapshot: %v", err) + } + if !strings.Contains(string(runtimePayload), endpoint.ID) || strings.Contains(string(runtimePayload), session.SessionToken) { + t.Fatalf("unexpected runtime snapshot payload: %s", runtimePayload) + } + + reloaded, err := NewFileStore(path) + if err != nil { + t.Fatalf("reload file store: %v", err) + } + if got, err := reloaded.RunEndpoints().Get(endpoint.ID); err != nil || got.Version != endpoint.Version { + t.Fatalf("runtime endpoint did not reload: %+v err=%v", got, err) + } + if got, err := reloaded.RunControlSessions().Get(endpoint.ID); err != nil || got.SessionTokenHash != session.SessionTokenHash || got.UsedNonces[0] != "nonce-1" { + t.Fatalf("runtime session did not reload: %+v err=%v", got, err) + } +} + func TestFileStorePersistsPluginOperationsStateAcrossRestart(t *testing.T) { path := filepath.Join(t.TempDir(), "plugin-operations.json") store, err := NewFileStore(path) diff --git a/platform_web/api/client.test.ts b/platform_web/api/client.test.ts index 0db4123..d7afc5d 100644 --- a/platform_web/api/client.test.ts +++ b/platform_web/api/client.test.ts @@ -323,6 +323,9 @@ describe("PlatformApiClient AI providers", () => { if (url.endsWith("/api/v1/jobs?states=queued%2Crunning%2Cfailed")) { return jsonResponse({ items: [job], count: 1 }); } + if (url.endsWith("/api/v1/jobs?states=queued%2Crunning%2Cfailed&limit=25")) { + return jsonResponse({ items: [job], count: 1 }); + } if (url.endsWith("/api/v1/jobs?serverInstanceId=server-1&states=queued%2Crunning%2Cfailed")) { return jsonResponse({ items: [job], count: 1 }); } @@ -584,6 +587,7 @@ describe("PlatformApiClient AI providers", () => { await expect(client.listJobs()).resolves.toMatchObject({ count: 1 }); await expect(client.listJobs(server.id)).resolves.toMatchObject({ count: 1 }); await expect(client.listJobs(undefined, { states: ["queued", "running", "failed"] })).resolves.toMatchObject({ count: 1 }); + await expect(client.listJobs(undefined, { states: ["queued", "running", "failed"], limit: 25 })).resolves.toMatchObject({ count: 1 }); await expect(client.listJobs(server.id, { states: ["queued", "running", "failed"] })).resolves.toMatchObject({ count: 1 }); await expect(client.listArtifacts({ ownerKind: "job", ownerId: job.id, state: "available" })).resolves.toMatchObject({ count: 1, items: [{ id: artifact.id }] }); await expect(client.openArtifactDownload(artifact.id)).resolves.toMatchObject({ downloadUrl: "/api/v1/artifacts/artifact-1/content", rangeSupported: true }); @@ -625,7 +629,7 @@ describe("PlatformApiClient AI providers", () => { client.invokeAI({ requestId: "ai-1", serverInstanceId: server.id, purpose: "config.suggest", prompt: "Tune PVP safely", currentConfig: "server.name=Example Survival #1\n" }) ).resolves.toMatchObject({ status: "ok", usage: { mocked: true }, configRecommendation: { diffSummary: "review required" } }); - expect(fetchMock).toHaveBeenCalledTimes(46); + expect(fetchMock).toHaveBeenCalledTimes(47); }); it("normalizes server file workspace null arrays from older platform responses", async () => { diff --git a/platform_web/api/client.ts b/platform_web/api/client.ts index 456a187..970d4fd 100644 --- a/platform_web/api/client.ts +++ b/platform_web/api/client.ts @@ -152,6 +152,7 @@ interface JobListFilter { runEndpointId?: string; state?: JobState; states?: JobState[]; + limit?: number; } export class PlatformApiClient { @@ -925,6 +926,7 @@ function jobListQuery(serverInstanceId: string | undefined, filter: JobListFilte if (filter.runEndpointId) params.set("runEndpointId", filter.runEndpointId); if (filter.state) params.set("state", filter.state); if (filter.states?.length) params.set("states", filter.states.join(",")); + if (filter.limit !== undefined) params.set("limit", String(filter.limit)); const query = params.toString(); return query ? `?${query}` : ""; } diff --git a/platform_web/pages/HomePage.tsx b/platform_web/pages/HomePage.tsx index ad827f9..0578047 100644 --- a/platform_web/pages/HomePage.tsx +++ b/platform_web/pages/HomePage.tsx @@ -33,6 +33,9 @@ interface OverviewData { jobs: JobResponse[]; } +const overviewJobStates: JobResponse["state"][] = ["failed"]; +const overviewJobLimit = 50; + export interface HomePageInitialState { core?: OperationsModuleState; metrics?: OperationsModuleState; @@ -55,7 +58,7 @@ export function HomePage({ session, onNavigate, initialState }: HomePageProps) { try { const [instances, jobs] = await Promise.all([ platformApiClient.listServerInstances(), - platformApiClient.listJobs() + platformApiClient.listJobs(undefined, { states: overviewJobStates, limit: overviewJobLimit }) ]); setCore({ status: "ready", data: { instances: instances.items, jobs: jobs.items }, refreshedAt: refreshedNow() }); } catch (error) { diff --git a/platform_web/pages/MaintenancePage.tsx b/platform_web/pages/MaintenancePage.tsx index e81119e..23888ff 100644 --- a/platform_web/pages/MaintenancePage.tsx +++ b/platform_web/pages/MaintenancePage.tsx @@ -10,6 +10,9 @@ import { cx } from "../utils/classes"; type ModuleState = { status: "loading" } | { status: "error"; reason: string } | { status: "ready"; data: T }; +const maintenanceJobStates: JobResponse["state"][] = ["failed"]; +const maintenanceJobLimit = 100; + export function MaintenancePage({ session, operations, onNavigate }: PageComponentProps) { const [jobs, setJobs] = useState>({ status: "loading" }); const [servers, setServers] = useState>({ status: "loading" }); @@ -18,7 +21,7 @@ export function MaintenancePage({ session, operations, onNavigate }: PageCompone const refreshJobs = useCallback(async () => { setJobs({ status: "loading" }); try { - const response = await platformApiClient.listJobs(); + const response = await platformApiClient.listJobs(undefined, { states: maintenanceJobStates, limit: maintenanceJobLimit }); setJobs({ status: "ready", data: response.items }); } catch (error) { setJobs({ status: "error", reason: error instanceof Error ? error.message : "加载失败" }); diff --git a/platform_web/pages/ServersPage.tsx b/platform_web/pages/ServersPage.tsx index 9bc79a6..180a758 100644 --- a/platform_web/pages/ServersPage.tsx +++ b/platform_web/pages/ServersPage.tsx @@ -66,6 +66,7 @@ const serverListRefreshMs = 15000; const serverMetricFreshMs = 30000; const serverForceDeleteConfirmation = "FORCE DELETE"; const serverOperationalJobStates: JobResponse["state"][] = ["queued", "accepted", "running", "retrying", "failed"]; +const serverOperationalJobLimit = 200; export function ServersPage({ session, operations, onNavigate }: PageComponentProps) { const [listState, setListState] = useState("loading"); @@ -97,7 +98,7 @@ export function ServersPage({ session, operations, onNavigate }: PageComponentPr showLoading ? platformApiClient.listGamePlugins() : Promise.resolve(undefined), platformApiClient.listRunEndpoints({ status: "online" }), platformApiClient.listServerInstances(), - platformApiClient.listJobs(undefined, { states: serverOperationalJobStates }) + platformApiClient.listJobs(undefined, { states: serverOperationalJobStates, limit: serverOperationalJobLimit }) ]); if (pluginResponse) setPlugins(pluginResponse.items); setEndpoints(endpointResponse.items);