Reduce hot run polling pressure

2026-09-07 20:26:20 +08:00
parent 4ab7bb820d
commit fd9dbab669
16 changed files with 335 additions and 56 deletions
@@ -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)
@@ -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)
+}
@@ -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()
@@ -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)