// Package state runs the per-instance state-tracking pipeline: // - RCON poll loops that execute manifest-declared commands and parse the // output into AppStateUpdate messages (players online, etc.) // - log-line event matching that runs manifest-declared regexes against // each log line and emits PlayerEvent messages (join/leave/chat/death) // // A Tracker is owned by the dispatcher for the lifetime of a running // instance. It is stopped by cancelling the context passed to Run. package state import ( "context" "errors" "fmt" "log/slog" "net" "regexp" "strconv" "strings" "sync" "time" "google.golang.org/protobuf/types/known/timestamppb" "github.com/dbledeez/panel/agent/internal/rcon" modulepkg "github.com/dbledeez/panel/pkg/module" panelv1 "github.com/dbledeez/panel/proto/panel/v1" ) // Emitter is the upward path for tracker output. Implementations forward // messages onto the gRPC stream back to the Controller. type Emitter interface { EmitAppState(*panelv1.AppStateUpdate) EmitPlayerEvent(*panelv1.PlayerEvent) } // Config wires the tracker to an instance + its RCON endpoint. type Config struct { InstanceID string Manifest *modulepkg.Manifest RCONAddr string // "host:port" — required for tcp adapters (telnet, source_rcon) RCONPassword string // PasswordFunc, if set, overrides RCONPassword on each dial attempt. // Used when the module's entrypoint generates an RCON secret at first // boot (e.g. 7DTD's TelnetPassword in /game-saves/.panel-telnet-password). // Called with the dial context; the returned string is used as the // password for that single attempt. PasswordFunc func(ctx context.Context) (string, error) Emitter Emitter // ContainerID is the Docker container name/ID for the stdio adapter. // Ignored by tcp adapters. The stdio adapter calls Stdio.AttachStdio // with this id to get a stdin/stdout handle to the running game. ContainerID string // Stdio is the backend that knows how to attach to a container's // stdio. Required for stdio adapter, nil otherwise. Stdio rcon.StdioBackend // OnRconResult is an optional callback invoked after every RCON // exec attempt (poll loop or operator-driven). success=true means // the engine answered; success=false means the exec failed (dial, // timeout, dead conn). Used by the dispatcher's arkHangGuard to // count consecutive failures and trigger an auto-restart when the // Wine 9 listener wedges. Nil-safe: callers should check before invoke. OnRconResult func(success bool) } // Tracker is created at instance start time and run in a goroutine. type Tracker struct { log *slog.Logger cfg Config events []compiledEvent // stateRE holds the state-source parse regexes, compiled once at New so // a bad pattern surfaces as a construction error instead of silently // recompiling (and silently failing) on every poll tick. Keyed by the // pattern string — pollOnce receives the StateSource by value, so the // pattern is the stable identity. stateRE map[string]*regexp.Regexp mu sync.Mutex client rcon.Client } type compiledEvent struct { name string re *regexp.Regexp kind panelv1.PlayerEvent_Kind } // New constructs a Tracker. Event patterns are compiled now so config errors // surface before Run is called. func New(log *slog.Logger, cfg Config) (*Tracker, error) { if cfg.Manifest == nil { return nil, errors.New("manifest is required") } if cfg.Emitter == nil { return nil, errors.New("emitter is required") } events, err := compileEvents(cfg.Manifest.Events) if err != nil { return nil, err } // Compile state-source parse patterns up front — a bad pattern is a // manifest bug and should fail loudly at init, not degrade into a // per-poll recompile that silently yields no state updates. stateRE := map[string]*regexp.Regexp{} for i := range cfg.Manifest.StateSources { p := cfg.Manifest.StateSources[i].Parse if p == nil || p.Kind != "regex" || p.Pattern == "" { continue } re, err := regexp.Compile(p.Pattern) if err != nil { return nil, fmt.Errorf("state source %d: compile pattern: %w", i, err) } stateRE[p.Pattern] = re } return &Tracker{ log: log.With("instance_id", cfg.InstanceID, "component", "state"), cfg: cfg, events: events, stateRE: stateRE, }, nil } // Client returns the currently-connected RCON client, or nil if the tracker // hasn't finished dialing (or the manifest has no RCON at all). Used by the // dispatcher to handle ad-hoc RCON commands from operators. func (t *Tracker) Client() rcon.Client { t.mu.Lock() defer t.mu.Unlock() return t.client } // Exec runs one operator command via the tracker's cached client. On a dead // connection (EOF, broken pipe — Palworld closes RCON after 60s idle) we // redial once and retry, so operators don't see the 60s-idle EOF in the UI. // Returns ("", "not_connected", ...) if the tracker hasn't finished dialing. func (t *Tracker) Exec(ctx context.Context, cmd string) (string, error) { t.mu.Lock() client := t.client t.mu.Unlock() if client == nil { return "", errors.New("RCON not yet connected; try again in a moment") } execCtx, cancel := context.WithTimeout(ctx, 10*time.Second) defer cancel() out, err := client.Exec(execCtx, cmd) if err == nil || !isDeadConnErr(err) { return out, err } t.redial(ctx) t.mu.Lock() client = t.client t.mu.Unlock() if client == nil { return out, err } retryCtx, retryCancel := context.WithTimeout(ctx, 10*time.Second) defer retryCancel() return client.Exec(retryCtx, cmd) } // OnLogLine runs each log line through compiled event patterns and emits a // PlayerEvent on the first match. Called from the dispatcher's log pump; must // be fast and non-blocking. func (t *Tracker) OnLogLine(line string) { for _, ev := range t.events { m := ev.re.FindStringSubmatch(line) if m == nil { continue } name := firstGroup(ev.re, m, "name", "player_name") id := firstGroup(ev.re, m, "platform_id", "owner", "cross_id", "id", "player_id") detail := firstGroup(ev.re, m, "msg", "detail", "reason") t.cfg.Emitter.EmitPlayerEvent(&panelv1.PlayerEvent{ InstanceId: t.cfg.InstanceID, Kind: ev.kind, PlayerName: name, PlayerId: id, Detail: detail, At: timestamppb.Now(), }) return } } // Run connects to RCON (with retries) and starts one goroutine per declared // RCON state source. Returns when ctx is cancelled. // // If the manifest has no RCON config at all, Run is a no-op that blocks on // ctx.Done. If it has RCON but no state sources, we still dial once so // operator ad-hoc commands from the UI work — state-source poll loops just // don't spawn. func (t *Tracker) Run(ctx context.Context) { if t.cfg.Manifest.RCON == nil { <-ctx.Done() return } client, err := t.dialWithRetries(ctx) if err != nil { // Context cancelled before we could connect; nothing to do. return } t.mu.Lock() t.client = client t.mu.Unlock() defer func() { t.mu.Lock() defer t.mu.Unlock() if t.client != nil { _ = t.client.Close() t.client = nil } }() t.log.Info("rcon connected", "addr", t.cfg.RCONAddr, "adapter", t.cfg.Manifest.RCON.Adapter) var wg sync.WaitGroup for i := range t.cfg.Manifest.StateSources { ss := t.cfg.Manifest.StateSources[i] if ss.Type != "rcon" || ss.Every.Std() <= 0 { continue } wg.Add(1) go func() { defer wg.Done() t.runRCONPoller(ctx, ss) }() } // Block until the tracker is cancelled so the defer (which closes the // rcon client) fires on teardown, not immediately. With no rcon state // sources wg.Wait() returns instantly — without this extra wait the // client would vanish right after Run dialed, making ad-hoc commands // from the UI fail with "RCON not yet connected" forever. <-ctx.Done() wg.Wait() } func (t *Tracker) hasRCONStateSource() bool { for _, ss := range t.cfg.Manifest.StateSources { if ss.Type == "rcon" && ss.Every.Std() > 0 { return true } } return false } func (t *Tracker) dialWithRetries(ctx context.Context) (rcon.Client, error) { rc := t.cfg.Manifest.RCON dialer, err := rcon.DialerFor(rc.Adapter) if err != nil { return nil, err } // stdio has no host/port — it attaches to the running container's stdin. // We still retry-loop because the container may not be up yet when the // tracker starts (race between Start + activate). if rc.Adapter == "stdio" { backoff := 2 * time.Second for { dialCtx, cancel := context.WithTimeout(ctx, 10*time.Second) client, err := dialer.Dial(dialCtx, rcon.DialOptions{ ContainerID: t.cfg.ContainerID, Stdio: t.cfg.Stdio, }) cancel() if err == nil { return client, nil } t.log.Warn("stdio attach failed, retrying", "err", err, "backoff", backoff) select { case <-ctx.Done(): return nil, ctx.Err() case <-time.After(backoff): } if backoff < 30*time.Second { backoff *= 2 } } } host, portStr, err := net.SplitHostPort(t.cfg.RCONAddr) if err != nil { return nil, fmt.Errorf("rcon addr %q: %w", t.cfg.RCONAddr, err) } port, err := strconv.Atoi(portStr) if err != nil { return nil, fmt.Errorf("rcon port %q: %w", portStr, err) } // Dial-retry backoff capped at 8s (not 30s) — during the post-Start // warm-up window the server's RCON flips between responsive and slow // every few seconds, and a 32s ceiling meant one unlucky timeout // stranded the panel for half a minute before the next retry. Cap // at 8s so ragnarok/etc. get back onto RCON within a few seconds of // ASA actually being ready, without hammering the port on a truly // dead server (8s × a few retries is ~30s before the tracker gives // up per-cycle, which the agent's outer loop still reschedules). backoff := 2 * time.Second for { dialCtx, cancel := context.WithTimeout(ctx, 15*time.Second) // Resolve password per-attempt when PasswordFunc is set — the // module's entrypoint may be mid-generation on first boot, so re- // reading on each retry lets us pick up the final value as soon // as it's written. pw := t.cfg.RCONPassword if t.cfg.PasswordFunc != nil { if p, err := t.cfg.PasswordFunc(dialCtx); err == nil && p != "" { pw = p } } client, err := dialer.Dial(dialCtx, rcon.DialOptions{ Host: host, Port: port, Password: pw, ConnectTimeout: 5 * time.Second, ContainerID: t.cfg.ContainerID, }) cancel() if err == nil { return client, nil } t.log.Warn("rcon dial failed, retrying", "err", err, "backoff", backoff) select { case <-ctx.Done(): return nil, ctx.Err() case <-time.After(backoff): } if backoff < 8*time.Second { backoff *= 2 } } } func (t *Tracker) runRCONPoller(ctx context.Context, ss modulepkg.StateSource) { every := ss.Every.Std() ticker := time.NewTicker(every) defer ticker.Stop() t.pollOnce(ctx, ss) // immediate first tick for { select { case <-ctx.Done(): return case <-ticker.C: t.pollOnce(ctx, ss) } } } func (t *Tracker) pollOnce(ctx context.Context, ss modulepkg.StateSource) { t.mu.Lock() client := t.client t.mu.Unlock() if client == nil { return } execCtx, cancel := context.WithTimeout(ctx, 10*time.Second) defer cancel() out, err := client.Exec(execCtx, ss.Command) if err != nil { t.log.Warn("rcon exec failed", "cmd", ss.Command, "err", err) if t.cfg.OnRconResult != nil { t.cfg.OnRconResult(false) } if isDeadConnErr(err) { t.redial(ctx) } return } if t.cfg.OnRconResult != nil { t.cfg.OnRconResult(true) } app := t.parseState(out, ss) if app == nil { // Parser yielded no structured fields (regex didn't match this game's // output, fields all empty, etc.) but the RCON exec succeeded — that // alone is proof the server is responsive. Emit a bare update keyed // by InstanceID so the dispatcher can use it to promote a stale // CRASHED status back to RUNNING. app = &panelv1.AppStateUpdate{InstanceId: t.cfg.InstanceID} } t.cfg.Emitter.EmitAppState(app) } // isDeadConnErr returns true for errors that indicate the RCON socket is no // longer usable (EOF, connection reset, broken pipe). Palworld's RCON, for // instance, closes the socket after ~60s idle — any subsequent Exec fails // with EOF and stays broken until we redial. func isDeadConnErr(err error) bool { if err == nil { return false } s := err.Error() return strings.Contains(s, "EOF") || strings.Contains(s, "broken pipe") || strings.Contains(s, "connection reset") || strings.Contains(s, "forcibly closed") || strings.Contains(s, "use of closed network connection") } // redial closes the current client and dials a fresh one. Runs inline with // the poll loop so the next tick gets a working client. On failure we leave // t.client = nil and the next poll will call us again. func (t *Tracker) redial(ctx context.Context) { t.mu.Lock() if t.client != nil { _ = t.client.Close() t.client = nil } t.mu.Unlock() t.log.Info("rcon redialing after dead connection") fresh, err := t.dialWithRetries(ctx) if err != nil { t.log.Warn("rcon redial failed", "err", err) return } t.mu.Lock() t.client = fresh t.mu.Unlock() t.log.Info("rcon reconnected", "addr", t.cfg.RCONAddr, "adapter", t.cfg.Manifest.RCON.Adapter) } // parseState runs the configured parser against raw RCON output using the // regex precompiled at New, returning an AppStateUpdate or nil if nothing // matched. func (t *Tracker) parseState(out string, ss modulepkg.StateSource) *panelv1.AppStateUpdate { if ss.Parse == nil || ss.Parse.Kind != "regex" { return nil } re := t.stateRE[ss.Parse.Pattern] if re == nil { return nil } return parseRegexCompiled(t.cfg.InstanceID, out, re, ss.Parse) } // parseStateOutput is the uncached variant (compiles the pattern per call). // Kept for tests and out-of-tracker callers; the tracker's poll loop uses // the precompiled parseState path. func parseStateOutput(instanceID, out string, ss modulepkg.StateSource) *panelv1.AppStateUpdate { if ss.Parse == nil { return nil } switch ss.Parse.Kind { case "regex": return parseRegex(instanceID, out, ss.Parse) default: return nil } } func parseRegex(instanceID, out string, p *modulepkg.ParseConfig) *panelv1.AppStateUpdate { re, err := regexp.Compile(p.Pattern) if err != nil { return nil } return parseRegexCompiled(instanceID, out, re, p) } func parseRegexCompiled(instanceID, out string, re *regexp.Regexp, p *modulepkg.ParseConfig) *panelv1.AppStateUpdate { m := re.FindStringSubmatch(out) if m == nil { return nil } app := &panelv1.AppStateUpdate{ InstanceId: instanceID, At: timestamppb.Now(), } for outName, groupName := range p.Fields { val := firstGroup(re, m, groupName) switch outName { case "players_online": if n, err := strconv.Atoi(val); err == nil { app.PlayersOnline = int32(n) } case "players_max": if n, err := strconv.Atoi(val); err == nil { app.PlayersMax = int32(n) } case "uptime_seconds": if n, err := strconv.ParseInt(val, 10, 64); err == nil { app.UptimeSeconds = n } default: if app.ModuleFields == nil { app.ModuleFields = map[string]string{} } app.ModuleFields[outName] = val } } return app } func compileEvents(events map[string]modulepkg.Event) ([]compiledEvent, error) { out := make([]compiledEvent, 0, len(events)) for name, ev := range events { re, err := regexp.Compile(ev.Pattern) if err != nil { return nil, fmt.Errorf("event %q: compile pattern: %w", name, err) } out = append(out, compiledEvent{ name: name, re: re, kind: parseKind(ev.Kind), }) } return out, nil } func parseKind(s string) panelv1.PlayerEvent_Kind { switch s { case "join": return panelv1.PlayerEvent_KIND_JOIN case "leave": return panelv1.PlayerEvent_KIND_LEAVE case "chat": return panelv1.PlayerEvent_KIND_CHAT case "death": return panelv1.PlayerEvent_KIND_DEATH default: return panelv1.PlayerEvent_KIND_CUSTOM } } // firstGroup returns the first non-empty named submatch from names. Missing // or empty groups are treated as absent so we try the next name. func firstGroup(re *regexp.Regexp, m []string, names ...string) string { for _, name := range names { i := re.SubexpIndex(name) if i >= 0 && i < len(m) && m[i] != "" { return m[i] } } return "" }