feat(scum-companion): add bounded dispatcher runtime
This commit is contained in:
npc0-hue
@@ -14,7 +14,7 @@
|
||||
|
||||
## 3. Implement the long-running SCUM Companion runtime
|
||||
|
||||
- [ ] 3.1 Implement authenticated registration, bounded dispatch polling, acknowledgement, idempotent result completion, backoff, and typed diagnostics in the SCUM Companion.
|
||||
- [x] 3.1 Implement authenticated registration, bounded dispatch polling, acknowledgement, idempotent result completion, backoff, and typed diagnostics in the SCUM Companion.
|
||||
- [ ] 3.2 Add a handler registry that validates declared schema, bound server, approval, server version, capability discovery, expiry, and idempotency before invoking an adapter.
|
||||
- [ ] 3.3 Implement safe configuration read/patch and diagnostics adapters that use only platform-authorized channels and redact host paths, credentials, and raw command text.
|
||||
- [ ] 3.4 Add Companion integration tests for command claiming, duplicate delivery, cancellation/expiry, malformed payloads, unsupported versions, and redaction.
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
@@ -18,53 +19,252 @@ type SafeAdapter interface {
|
||||
NotifyPlayer(context.Context, map[string]any) (map[string]any, error)
|
||||
}
|
||||
|
||||
type HandlerAvailability struct { ServerVersion string; Capabilities map[string]bool; Approved bool }
|
||||
type HandlerAvailability struct {
|
||||
ServerVersion string
|
||||
Capabilities map[string]bool
|
||||
Approved bool
|
||||
}
|
||||
type CommandHandler func(context.Context, map[string]any) (map[string]any, error)
|
||||
type HandlerRegistry struct { availability HandlerAvailability; handlers map[string]CommandHandler }
|
||||
type HandlerRegistry struct {
|
||||
availability HandlerAvailability
|
||||
handlers map[string]CommandHandler
|
||||
mu sync.Mutex
|
||||
completed map[string]CommandResult
|
||||
}
|
||||
|
||||
func NewHandlerRegistry(availability HandlerAvailability, adapter SafeAdapter) *HandlerRegistry {
|
||||
registry := &HandlerRegistry{availability: availability, handlers: map[string]CommandHandler{}}
|
||||
if adapter == nil { return registry }
|
||||
registry.handlers["config.read"] = func(ctx context.Context, _ map[string]any) (map[string]any, error) { return adapter.ReadConfiguration(ctx) }
|
||||
registry.handlers["config.patch"] = func(ctx context.Context, payload map[string]any) (map[string]any, error) { return adapter.PatchConfiguration(ctx, payload) }
|
||||
registry := &HandlerRegistry{availability: availability, handlers: map[string]CommandHandler{}, completed: map[string]CommandResult{}}
|
||||
if adapter == nil {
|
||||
return registry
|
||||
}
|
||||
registry.handlers["config.read"] = func(ctx context.Context, _ map[string]any) (map[string]any, error) {
|
||||
return adapter.ReadConfiguration(ctx)
|
||||
}
|
||||
registry.handlers["config.patch"] = func(ctx context.Context, payload map[string]any) (map[string]any, error) {
|
||||
return adapter.PatchConfiguration(ctx, payload)
|
||||
}
|
||||
registry.handlers["companion.diagnostics"] = func(ctx context.Context, _ map[string]any) (map[string]any, error) { return adapter.Diagnostics(ctx) }
|
||||
registry.handlers["game-state.patch"] = func(ctx context.Context, payload map[string]any) (map[string]any, error) { return adapter.PatchGameState(ctx, payload) }
|
||||
registry.handlers["reward.deliver"] = func(ctx context.Context, payload map[string]any) (map[string]any, error) { return adapter.DeliverReward(ctx, payload) }
|
||||
registry.handlers["player.notify"] = func(ctx context.Context, payload map[string]any) (map[string]any, error) { return adapter.NotifyPlayer(ctx, payload) }
|
||||
registry.handlers["game-state.patch"] = func(ctx context.Context, payload map[string]any) (map[string]any, error) {
|
||||
return adapter.PatchGameState(ctx, payload)
|
||||
}
|
||||
registry.handlers["reward.deliver"] = func(ctx context.Context, payload map[string]any) (map[string]any, error) {
|
||||
return adapter.DeliverReward(ctx, payload)
|
||||
}
|
||||
registry.handlers["player.notify"] = func(ctx context.Context, payload map[string]any) (map[string]any, error) {
|
||||
return adapter.NotifyPlayer(ctx, payload)
|
||||
}
|
||||
return registry
|
||||
}
|
||||
|
||||
func (registry *HandlerRegistry) Execute(ctx context.Context, command ClaimedCommand) (CommandResult, error) {
|
||||
if err := validateDeclaredCommand(command); err != nil { return unsupportedResult("validation-failed"), nil }
|
||||
if !registry.availability.Approved || !registry.availability.Capabilities[command.CommandType] { return unsupportedResult("unsupported"), nil }
|
||||
registry.mu.Lock()
|
||||
cached, done := registry.completed[command.ID]
|
||||
registry.mu.Unlock()
|
||||
if done {
|
||||
return cached, nil
|
||||
}
|
||||
if err := validateDeclaredCommandAt(command, time.Now); err != nil {
|
||||
return unsupportedResult("validation-failed"), nil
|
||||
}
|
||||
if !registry.availability.Approved || !registry.availability.Capabilities[command.CommandType] {
|
||||
return unsupportedResult("unsupported"), nil
|
||||
}
|
||||
handler, exists := registry.handlers[command.CommandType]
|
||||
if !exists || strings.TrimSpace(registry.availability.ServerVersion) == "" { return unsupportedResult("unsupported"), nil }
|
||||
if !exists || strings.TrimSpace(registry.availability.ServerVersion) == "" {
|
||||
return unsupportedResult("unsupported"), nil
|
||||
}
|
||||
payload, err := handler(ctx, command.Payload)
|
||||
if err != nil { return CommandResult{Status: "failed", Summary: "typed adapter failed", Payload: map[string]any{"result": "failed"}}, nil }
|
||||
return CommandResult{Status: "succeeded", Summary: "typed adapter completed", Payload: redactTypedPayload(payload)}, nil
|
||||
if err != nil {
|
||||
return CommandResult{Status: "failed", Summary: "typed adapter failed", Payload: map[string]any{"result": "failed"}}, nil
|
||||
}
|
||||
result := CommandResult{Status: "succeeded", Summary: "typed adapter completed", Payload: redactTypedPayload(payload)}
|
||||
registry.mu.Lock()
|
||||
registry.completed[command.ID] = result
|
||||
registry.mu.Unlock()
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func validateDeclaredCommand(command ClaimedCommand) error {
|
||||
if command.ID == "" || command.ProfileKey != ProfileKey || command.FencingToken == 0 || command.Payload == nil || command.ExpiresAt.IsZero() || !time.Now().Before(command.ExpiresAt) { return fmt.Errorf("invalid command") }
|
||||
for key, value := range command.Payload { if !safeCommandField(key, value) { return fmt.Errorf("unsafe payload") } }
|
||||
func validateDeclaredCommandAt(command ClaimedCommand, now func() time.Time) error {
|
||||
if command.ID == "" || command.ProfileKey != ProfileKey || command.FencingToken == 0 || command.Payload == nil || command.LeaseExpiresAt.IsZero() || command.ExpiresAt.IsZero() || !now().Before(command.LeaseExpiresAt) || !now().Before(command.ExpiresAt) {
|
||||
return fmt.Errorf("invalid command")
|
||||
}
|
||||
for key, value := range command.Payload {
|
||||
if !safeCommandField(key, value) {
|
||||
return fmt.Errorf("unsafe payload")
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func safeCommandField(key string, value any) bool {
|
||||
lower := strings.ToLower(strings.TrimSpace(key)); if lower == "" || strings.Contains(lower, "path") || strings.Contains(lower, "credential") || strings.Contains(lower, "password") || strings.Contains(lower, "sql") || strings.Contains(lower, "rcon") || strings.Contains(lower, "command") { return false }
|
||||
if text, ok := value.(string); ok { compact := strings.ToLower(text); return !strings.Contains(compact, "bearer ") && !strings.Contains(compact, "password=") && !strings.Contains(compact, "select ") && !strings.Contains(compact, "/users/") }
|
||||
lower := strings.ToLower(strings.TrimSpace(key))
|
||||
if lower == "" || strings.Contains(lower, "path") || strings.Contains(lower, "credential") || strings.Contains(lower, "password") || strings.Contains(lower, "sql") || strings.Contains(lower, "rcon") || strings.Contains(lower, "command") {
|
||||
return false
|
||||
}
|
||||
if text, ok := value.(string); ok {
|
||||
compact := strings.ToLower(text)
|
||||
return !strings.Contains(compact, "bearer ") && !strings.Contains(compact, "password=") && !strings.Contains(compact, "select ") && !strings.Contains(compact, "/users/")
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func unsupportedResult(code string) CommandResult { return CommandResult{Status: "failed", Summary: "typed operation unavailable", Payload: map[string]any{"result": code}} }
|
||||
func redactTypedPayload(payload map[string]any) map[string]any { result := map[string]any{}; for key, value := range payload { if safeCommandField(key, value) { result[key] = value } }; return result }
|
||||
func unsupportedResult(code string) CommandResult {
|
||||
return CommandResult{Status: "failed", Summary: "typed operation unavailable", Payload: map[string]any{"result": code}}
|
||||
}
|
||||
func redactTypedPayload(payload map[string]any) map[string]any {
|
||||
result := map[string]any{}
|
||||
for key, value := range payload {
|
||||
if safeCommandField(key, value) {
|
||||
result[key] = value
|
||||
}
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
type CommandGateway interface {
|
||||
ClaimCommands(context.Context, int) ([]ClaimedCommand, error)
|
||||
AckCommand(context.Context, string, uint64) (CommandAck, error)
|
||||
CompleteCommand(context.Context, string, uint64, CommandResult) (CompletedCommand, error)
|
||||
}
|
||||
type Dispatcher struct {
|
||||
Client CommandGateway
|
||||
Registry *HandlerRegistry
|
||||
PollLimit int
|
||||
Backoff time.Duration
|
||||
Now func() time.Time
|
||||
}
|
||||
|
||||
type Dispatcher struct { Client *Client; Registry *HandlerRegistry; PollLimit int; Backoff time.Duration }
|
||||
func (dispatcher Dispatcher) DispatchOnce(ctx context.Context) error {
|
||||
if dispatcher.Client == nil || dispatcher.Registry == nil { return fmt.Errorf("dispatcher is not configured") }
|
||||
limit := dispatcher.PollLimit; if limit == 0 { limit = 10 }; if limit < 1 || limit > 50 { return fmt.Errorf("dispatcher poll limit is invalid") }
|
||||
commands, err := dispatcher.Client.ClaimCommands(ctx, limit); if err != nil { return err }
|
||||
for _, command := range commands { if _, err = dispatcher.Client.AckCommand(ctx, command.ID, command.FencingToken); err != nil { return err }; result, executionErr := dispatcher.Registry.Execute(ctx, command); if executionErr != nil { result = CommandResult{Status: "failed", Summary: "typed adapter failed", Payload: map[string]any{"result": "failed"}} }; if _, err = dispatcher.Client.CompleteCommand(ctx, command.ID, command.FencingToken, result); err != nil { return err } }
|
||||
if dispatcher.Client == nil || dispatcher.Registry == nil {
|
||||
return fmt.Errorf("dispatcher is not configured")
|
||||
}
|
||||
limit := dispatcher.PollLimit
|
||||
if limit == 0 {
|
||||
limit = 10
|
||||
}
|
||||
if limit < 1 || limit > 50 {
|
||||
return fmt.Errorf("dispatcher poll limit is invalid")
|
||||
}
|
||||
commands, err := dispatcher.Client.ClaimCommands(ctx, limit)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, command := range commands {
|
||||
if err := validateDeclaredCommandAt(command, dispatcher.clock()); err != nil {
|
||||
if completeErr := dispatcher.completeValidationFailure(ctx, command); completeErr != nil {
|
||||
return completeErr
|
||||
}
|
||||
continue
|
||||
}
|
||||
if _, err = dispatcher.Client.AckCommand(ctx, command.ID, command.FencingToken); err != nil {
|
||||
return err
|
||||
}
|
||||
result, executionErr := dispatcher.Registry.Execute(ctx, command)
|
||||
if executionErr != nil {
|
||||
result = CommandResult{Status: "failed", Summary: "typed adapter failed", Payload: map[string]any{"result": "failed"}}
|
||||
}
|
||||
if _, err = dispatcher.Client.CompleteCommand(ctx, command.ID, command.FencingToken, result); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
func (dispatcher Dispatcher) Run(ctx context.Context) error { backoff := dispatcher.Backoff; if backoff <= 0 { backoff = 2 * time.Second }; for { if err := dispatcher.DispatchOnce(ctx); err != nil { select { case <-ctx.Done(): return ctx.Err(); case <-time.After(backoff): continue } }; select { case <-ctx.Done(): return ctx.Err(); case <-time.After(backoff): } } }
|
||||
func (dispatcher Dispatcher) completeValidationFailure(ctx context.Context, command ClaimedCommand) error {
|
||||
if command.ID == "" || command.FencingToken == 0 {
|
||||
return fmt.Errorf("claimed command is invalid")
|
||||
}
|
||||
_, err := dispatcher.Client.CompleteCommand(ctx, command.ID, command.FencingToken, unsupportedResult("validation-failed"))
|
||||
return err
|
||||
}
|
||||
func (dispatcher Dispatcher) clock() func() time.Time {
|
||||
if dispatcher.Now != nil {
|
||||
return dispatcher.Now
|
||||
}
|
||||
return time.Now
|
||||
}
|
||||
func (dispatcher Dispatcher) Run(ctx context.Context) error {
|
||||
backoff := dispatcher.Backoff
|
||||
if backoff <= 0 {
|
||||
backoff = 2 * time.Second
|
||||
}
|
||||
for {
|
||||
if err := dispatcher.DispatchOnce(ctx); err != nil {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-time.After(backoff):
|
||||
continue
|
||||
}
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-time.After(backoff):
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Runtime keeps the registered companion alive with bounded heartbeat and
|
||||
// polling intervals. It owns no host connection or game credential; handlers
|
||||
// are the only route to a version-bound adapter.
|
||||
type RuntimeGateway interface {
|
||||
CommandGateway
|
||||
Register(context.Context) (Registration, error)
|
||||
Heartbeat(context.Context, HealthReport) (HeartbeatResult, error)
|
||||
}
|
||||
type Runtime struct {
|
||||
Client RuntimeGateway
|
||||
Dispatcher Dispatcher
|
||||
HeartbeatEvery time.Duration
|
||||
PollEvery time.Duration
|
||||
Backoff time.Duration
|
||||
Health func() HealthReport
|
||||
}
|
||||
|
||||
func (runtime Runtime) Run(ctx context.Context) error {
|
||||
if runtime.Client == nil || runtime.Dispatcher.Registry == nil {
|
||||
return fmt.Errorf("runtime is not configured")
|
||||
}
|
||||
if _, err := runtime.Client.Register(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
heartbeatEvery, pollEvery := runtime.HeartbeatEvery, runtime.PollEvery
|
||||
if heartbeatEvery < 5*time.Second {
|
||||
heartbeatEvery = 30 * time.Second
|
||||
}
|
||||
if pollEvery < time.Second {
|
||||
pollEvery = 5 * time.Second
|
||||
}
|
||||
heartbeat := time.NewTicker(heartbeatEvery)
|
||||
defer heartbeat.Stop()
|
||||
poll := time.NewTicker(pollEvery)
|
||||
defer poll.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-heartbeat.C:
|
||||
report := HealthReport{Status: "healthy", Reason: "typed companion dispatcher ready"}
|
||||
if runtime.Health != nil {
|
||||
report = runtime.Health()
|
||||
}
|
||||
if _, err := runtime.Client.Heartbeat(ctx, report); err != nil {
|
||||
return err
|
||||
}
|
||||
case <-poll.C:
|
||||
if err := runtime.Dispatcher.DispatchOnce(ctx); err != nil {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-time.After(runtime.backoff()):
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
func (runtime Runtime) backoff() time.Duration {
|
||||
if runtime.Backoff > 0 {
|
||||
return runtime.Backoff
|
||||
}
|
||||
return 2 * time.Second
|
||||
}
|
||||
|
||||
@@ -0,0 +1,80 @@
|
||||
package companion
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
type dispatchFixture struct {
|
||||
commands []ClaimedCommand
|
||||
acks []string
|
||||
completed []CommandResult
|
||||
}
|
||||
|
||||
func (fixture *dispatchFixture) ClaimCommands(_ context.Context, _ int) ([]ClaimedCommand, error) {
|
||||
return append([]ClaimedCommand(nil), fixture.commands...), nil
|
||||
}
|
||||
func (fixture *dispatchFixture) AckCommand(_ context.Context, id string, _ uint64) (CommandAck, error) {
|
||||
fixture.acks = append(fixture.acks, id)
|
||||
return CommandAck{CommandID: id, State: "claimed", FencingToken: 7}, nil
|
||||
}
|
||||
func (fixture *dispatchFixture) CompleteCommand(_ context.Context, _ string, _ uint64, result CommandResult) (CompletedCommand, error) {
|
||||
fixture.completed = append(fixture.completed, result)
|
||||
return CompletedCommand{State: "succeeded"}, nil
|
||||
}
|
||||
|
||||
type adapterFixture struct{ reads int }
|
||||
|
||||
func (adapter *adapterFixture) ReadConfiguration(context.Context) (map[string]any, error) {
|
||||
adapter.reads++
|
||||
return map[string]any{"version": "0.9.700.90357", "hostPath": "C:/must-redact"}, nil
|
||||
}
|
||||
func (*adapterFixture) PatchConfiguration(context.Context, map[string]any) (map[string]any, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (*adapterFixture) Diagnostics(context.Context) (map[string]any, error) {
|
||||
return map[string]any{"status": "healthy"}, nil
|
||||
}
|
||||
func (*adapterFixture) PatchGameState(context.Context, map[string]any) (map[string]any, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (*adapterFixture) DeliverReward(context.Context, map[string]any) (map[string]any, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (*adapterFixture) NotifyPlayer(context.Context, map[string]any) (map[string]any, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func TestDispatcherAcknowledgesOnlyLiveValidatedTypedCommands(t *testing.T) {
|
||||
stamp := time.Now().UTC()
|
||||
adapter := &adapterFixture{}
|
||||
registry := NewHandlerRegistry(HandlerAvailability{ServerVersion: "0.9.700.90357", Approved: true, Capabilities: map[string]bool{"config.read": true}}, adapter)
|
||||
fixture := &dispatchFixture{commands: []ClaimedCommand{{ID: "read-1", ProfileKey: ProfileKey, CommandType: "config.read", Payload: map[string]any{}, FencingToken: 7, LeaseExpiresAt: stamp.Add(time.Minute), ExpiresAt: stamp.Add(time.Minute)}, {ID: "expired-1", ProfileKey: ProfileKey, CommandType: "config.read", Payload: map[string]any{}, FencingToken: 8, LeaseExpiresAt: stamp.Add(-time.Second), ExpiresAt: stamp.Add(-time.Second)}}}
|
||||
dispatcher := Dispatcher{Client: fixture, Registry: registry, Now: func() time.Time { return stamp }}
|
||||
if err := dispatcher.DispatchOnce(context.Background()); err != nil {
|
||||
t.Fatalf("dispatch: %v", err)
|
||||
}
|
||||
if len(fixture.acks) != 1 || fixture.acks[0] != "read-1" || len(fixture.completed) != 2 || adapter.reads != 1 {
|
||||
t.Fatalf("unexpected bounded dispatch: acks=%v completed=%+v reads=%d", fixture.acks, fixture.completed, adapter.reads)
|
||||
}
|
||||
if fixture.completed[0].Payload["hostPath"] != nil || fixture.completed[1].Payload["result"] != "validation-failed" {
|
||||
t.Fatalf("unsafe or malformed result: %+v", fixture.completed)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRegistryReturnsCachedResultForDuplicateDelivery(t *testing.T) {
|
||||
stamp := time.Now().UTC()
|
||||
adapter := &adapterFixture{}
|
||||
registry := NewHandlerRegistry(HandlerAvailability{ServerVersion: "0.9.700.90357", Approved: true, Capabilities: map[string]bool{"config.read": true}}, adapter)
|
||||
command := ClaimedCommand{ID: "duplicate-1", ProfileKey: ProfileKey, CommandType: "config.read", Payload: map[string]any{}, FencingToken: 7, LeaseExpiresAt: stamp.Add(time.Minute), ExpiresAt: stamp.Add(time.Minute)}
|
||||
if _, err := registry.Execute(context.Background(), command); err != nil {
|
||||
t.Fatalf("first execute: %v", err)
|
||||
}
|
||||
if _, err := registry.Execute(context.Background(), command); err != nil {
|
||||
t.Fatalf("second execute: %v", err)
|
||||
}
|
||||
if adapter.reads != 1 {
|
||||
t.Fatalf("duplicate delivery invoked adapter %d times", adapter.reads)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user