From 8c924b5928a17798fe0527916acf098207d1c3e7 Mon Sep 17 00:00:00 2001 From: npc0-hue Date: Wed, 9 Sep 2026 16:21:12 +0800 Subject: [PATCH] Reduce log ingest and SCUM page refresh pressure --- platform/repo/file_store.go | 5 ++++- platform/repo/mysql_store.go | 4 +++- platform/repo/resources_test.go | 9 ++++++++- platform/service/resources.go | 18 ++++++++++++++---- .../scum-server-plugin/features/page-data.ts | 2 +- .../scum-server-plugin/features/page.ts | 5 +++-- plugins/tests/scum-feature-module.test.ts | 11 ++++++++++- 7 files changed, 43 insertions(+), 11 deletions(-) diff --git a/platform/repo/file_store.go b/platform/repo/file_store.go index f5a4d30..6b4daf2 100644 --- a/platform/repo/file_store.go +++ b/platform/repo/file_store.go @@ -41,6 +41,7 @@ type StoreSnapshot struct { type runtimeSnapshot struct { RunControlSessions []domain.RunControlSession `json:"runControlSessions"` RunEndpoints []domain.RunEndpoint `json:"runEndpoints"` + LogStreams []domain.LogStream `json:"logStreams"` } type FileStore struct { @@ -131,7 +132,7 @@ func (store *FileStore) RunUpdateJobs() RunUpdateJobRepository { } func (store *FileStore) LogStreams() LogStreamRepository { - return &persistentRepository[domain.LogStream, domain.LogStreamFilter]{repository: store.MemoryStore.logStreams, persist: store.persist} + return &persistentRepository[domain.LogStream, domain.LogStreamFilter]{repository: store.MemoryStore.logStreams, persist: store.persistRuntime} } func (store *FileStore) MetricSamples() MetricSampleRepository { @@ -318,6 +319,7 @@ func (store *FileStore) runtimeSnapshot() runtimeSnapshot { return runtimeSnapshot{ RunControlSessions: snapshotRepository(store.MemoryStore.runSessions), RunEndpoints: snapshotRepository(store.MemoryStore.runEndpoints), + LogStreams: snapshotRepository(store.MemoryStore.logStreams), } } @@ -350,6 +352,7 @@ func (store *FileStore) loadSnapshot(snapshot StoreSnapshot) { func (store *FileStore) loadRuntimeSnapshot(snapshot runtimeSnapshot) { loadRepository(store.MemoryStore.runSessions, snapshot.RunControlSessions) loadRepository(store.MemoryStore.runEndpoints, snapshot.RunEndpoints) + loadRepository(store.MemoryStore.logStreams, snapshot.LogStreams) } type mutableRepository[T any, F any] interface { diff --git a/platform/repo/mysql_store.go b/platform/repo/mysql_store.go index 2f4d8a3..b83f176 100644 --- a/platform/repo/mysql_store.go +++ b/platform/repo/mysql_store.go @@ -114,7 +114,7 @@ func (store *MySQLStore) RunUpdateJobs() RunUpdateJobRepository { } func (store *MySQLStore) LogStreams() LogStreamRepository { - return &persistentRepository[domain.LogStream, domain.LogStreamFilter]{repository: store.MemoryStore.logStreams, persist: store.persist} + return &persistentRepository[domain.LogStream, domain.LogStreamFilter]{repository: store.MemoryStore.logStreams, persist: store.persistRuntime} } func (store *MySQLStore) MetricSamples() MetricSampleRepository { @@ -280,6 +280,7 @@ func (store *MySQLStore) runtimeSnapshot() runtimeSnapshot { return runtimeSnapshot{ RunControlSessions: snapshotRepository(store.MemoryStore.runSessions), RunEndpoints: snapshotRepository(store.MemoryStore.runEndpoints), + LogStreams: snapshotRepository(store.MemoryStore.logStreams), } } @@ -312,4 +313,5 @@ func (store *MySQLStore) loadSnapshot(snapshot StoreSnapshot) { func (store *MySQLStore) loadRuntimeSnapshot(snapshot runtimeSnapshot) { loadRepository(store.MemoryStore.runSessions, snapshot.RunControlSessions) loadRepository(store.MemoryStore.runEndpoints, snapshot.RunEndpoints) + loadRepository(store.MemoryStore.logStreams, snapshot.LogStreams) } diff --git a/platform/repo/resources_test.go b/platform/repo/resources_test.go index d2eb43f..0805bfb 100644 --- a/platform/repo/resources_test.go +++ b/platform/repo/resources_test.go @@ -330,6 +330,10 @@ func TestFileStorePersistsRunRuntimeStateSeparately(t *testing.T) { if err := store.RunControlSessions().Create(session); err != nil { t.Fatalf("create runtime session: %v", err) } + stream := domain.LogStream{ID: "run.run-runtime.server-1.scum.console.stdout", ServerInstanceID: "server-1", Source: domain.LogStreamSourceProcess, StreamKey: "scum.console.stdout", LatestSeq: 42, StorageBackend: domain.LogStorageBackendLocalSegments, RetentionPolicy: "default", CreatedAt: stamp, UpdatedAt: stamp} + if err := store.LogStreams().Create(stream); err != nil { + t.Fatalf("create runtime log stream: %v", err) + } mainAfter, err := os.ReadFile(path) if err != nil { t.Fatalf("read main snapshot after runtime update: %v", err) @@ -341,7 +345,7 @@ func TestFileStorePersistsRunRuntimeStateSeparately(t *testing.T) { if err != nil { t.Fatalf("read runtime snapshot: %v", err) } - if !strings.Contains(string(runtimePayload), endpoint.ID) || strings.Contains(string(runtimePayload), session.SessionToken) { + if !strings.Contains(string(runtimePayload), endpoint.ID) || !strings.Contains(string(runtimePayload), stream.ID) || strings.Contains(string(runtimePayload), session.SessionToken) { t.Fatalf("unexpected runtime snapshot payload: %s", runtimePayload) } @@ -355,6 +359,9 @@ func TestFileStorePersistsRunRuntimeStateSeparately(t *testing.T) { 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) } + if got, err := reloaded.LogStreams().Get(stream.ID); err != nil || got.LatestSeq != stream.LatestSeq { + t.Fatalf("runtime log stream did not reload: %+v err=%v", got, err) + } } func TestFileStorePersistsPluginOperationsStateAcrossRestart(t *testing.T) { diff --git a/platform/service/resources.go b/platform/service/resources.go index 62d7da5..62e8516 100644 --- a/platform/service/resources.go +++ b/platform/service/resources.go @@ -1619,12 +1619,22 @@ func (svc *CoreService) GetRunEndpoint(id string) (domain.RunEndpoint, error) { } func (svc *CoreService) ListRunEndpoints(filter domain.RunEndpointFilter) ([]domain.RunEndpoint, error) { - svc.controlMu.Lock() - defer svc.controlMu.Unlock() - if err := svc.sweepExpiredRunRegistrationsLocked(svc.now()); err != nil { + endpoints, err := svc.store.RunEndpoints().List(domain.RunEndpointFilter{}) + if err != nil { return nil, err } - return svc.store.RunEndpoints().List(filter) + stamp := svc.now() + items := make([]domain.RunEndpoint, 0, len(endpoints)) + for _, endpoint := range endpoints { + if !runEndpointRegistrationCurrentAt(endpoint, stamp) && (endpoint.Status == domain.RunEndpointStatusOnline || endpoint.Status == domain.RunEndpointStatusDegraded) { + endpoint.Status = domain.RunEndpointStatusOffline + } + if filter.Status != "" && endpoint.Status != filter.Status { + continue + } + items = append(items, domain.CopyRunEndpoint(endpoint)) + } + return items, nil } func (svc *CoreService) CreateServerInstance(instance domain.ServerInstance) (domain.ServerInstance, error) { diff --git a/plugins/examples/scum-server-plugin/features/page-data.ts b/plugins/examples/scum-server-plugin/features/page-data.ts index 320f664..4298b48 100644 --- a/plugins/examples/scum-server-plugin/features/page-data.ts +++ b/plugins/examples/scum-server-plugin/features/page-data.ts @@ -185,7 +185,7 @@ const queuedQueryBuckets = new Set(); export async function loadSCUMSurface(actions: SCUMWorkspaceActions, pageKey: string): Promise { if (!actions.pluginData) throw new Error("通用 pluginData 能力不可用。"); const canonical = canonicalPageKey(pageKey); - await queueSCUMDatabaseRefresh(actions, canonical).catch(() => undefined); + void queueSCUMDatabaseRefresh(actions, canonical).catch(() => undefined); const data: SCUMSurfaceData = { ...emptySCUMSurfaceData }; const keys = pageCollections[canonical]; const records = await Promise.all(keys.map(async (key) => [key, await actions.pluginData!.list(scumCollections[key])] as const)); diff --git a/plugins/examples/scum-server-plugin/features/page.ts b/plugins/examples/scum-server-plugin/features/page.ts index 141888f..6daeac5 100644 --- a/plugins/examples/scum-server-plugin/features/page.ts +++ b/plugins/examples/scum-server-plugin/features/page.ts @@ -32,6 +32,7 @@ type PlayerPanelState = { kind: PlayerPanelKind; playerId: string }; type AttributeDraft = { fieldKey: string; label: string; before: string; after: string }; const scumMapBackground = new URL("../assets/map/scum-map-overview.jpg", import.meta.url).href; const scumMapSize = 256; +const scumSurfaceRefreshMs = 15000; const rideDistanceThreshold = 50000; const vehicleIconByClass: Record = { BPC_Barba: new URL("../assets/vehicles/vehicle-BPC_Barba.webp", import.meta.url).href, @@ -141,14 +142,14 @@ export function renderSCUMFeaturePage(react: ReactLike, input: SCUMPageContext) if (react.useEffect) react.useEffect(() => { if (playerPanel.kind === "closed") refresh(); if (playerPanel.kind !== "closed") return; - const interval = setInterval(refresh, 5000); + const interval = setInterval(refresh, scumSurfaceRefreshMs); return () => clearInterval(interval); }, [input.serverInstanceId, pageKey, input.workspaceActions, playerPanel.kind]); const data = state.status === "ready" ? state.data : emptySCUMSurfaceData; return e("section", { className: "console-panel scum-workbench", "aria-label": input.pageTitle ?? surfaceTitle(pageKey) }, action.status !== "idle" ? e("p", { className: "page-status", "data-state": action.status }, action.message) : null, - state.status === "loading" ? e("p", { className: "page-status" }, "正在派发 SCUM 数据库模板查询并读取插件自有集合…") : null, + state.status === "loading" ? e("p", { className: "page-status" }, "正在读取平台本地 SCUM 投影，数据库查询会按节流入队…") : null, state.status === "error" ? e("p", { className: "page-status", "data-state": "error" }, state.reason) : null, state.status === "ready" ? renderSurfaceBody(e, pageKey, data, input, { playerSearch, setPlayerSearch, playerStatus, setPlayerStatus, playerPanel, setPlayerPanel, attributeDrafts, setAttributeDrafts, squadSearch, setSquadSearch, selectedSquadId, setSelectedSquadId, diff --git a/plugins/tests/scum-feature-module.test.ts b/plugins/tests/scum-feature-module.test.ts index 1834c2d..0a89c22 100644 --- a/plugins/tests/scum-feature-module.test.ts +++ b/plugins/tests/scum-feature-module.test.ts @@ -116,6 +116,14 @@ describe("SCUM plugin feature module", () => { expect(dataClientSource).not.toContain("logs.query"); }); + it("does not block page data reads on SCUM database refresh dispatch", async () => { + const releaseDispatches: Array<(value: { status: "queued" }) => void> = []; + const dispatch = vi.fn>(() => new Promise((resolve) => { releaseDispatches.push(resolve); })); + const list = vi.fn(async (collection: string) => ({ items: [{ key: `${collection}-1`, value: { collection } }], count: 1 })); + await expect(loadSCUMSurface({ pluginData: pluginDataActions({ list }), dispatch }, "live-map")).resolves.toMatchObject({ players: expect.any(Array), vehicles: expect.any(Array) }); + releaseDispatches.forEach((release) => release({ status: "queued" })); + }); + it("uses workflows as the manifest activity key and keeps activity as a compatibility alias", async () => { const list = vi.fn(async (collection: string) => ({ items: [{ key: `${collection}-1`, value: { collection } }], count: 1 })); await loadSCUMSurface({ pluginData: pluginDataActions({ list }) }, "workflows"); @@ -308,7 +316,8 @@ describe("SCUM plugin feature module", () => { expect(source).not.toContain("projectSCUMLoginLogs"); expect(source).not.toContain("logs.query"); expect(source).not.toContain("requestSCUMPageQueries"); - expect(pageSource).toContain("setInterval(refresh, 5000)"); + expect(pageSource).toContain("scumSurfaceRefreshMs = 15000"); + expect(pageSource).toContain("setInterval(refresh, scumSurfaceRefreshMs)"); expect(pageSource).toContain("clearInterval(interval)"); }); });