Documentation
¶
Overview ¶
Package busctl glues the consume client to the bus daemon. It encapsulates the three operations a consumer needs at startup and shutdown:
discover: find the running bus or fork a fresh one (race-free, plan §12
P3 "try dial → try fork lock → retry dial")
spawn: exec `dws event _bus --client-id <id>` as a detached
background process (stdio detach, setsid, ready pipe handshake)
stop: gracefully terminate the bus daemon (SIGTERM + IPC fallback)
All three operations are short-lived helpers — they own no long-lived goroutines and return whole errors to their caller.
Index ¶
- Constants
- Variables
- func Discover(cfg DiscoverConfig) (net.Conn, error)
- func LockPath(workDir string) string
- func MetaPath(workDir string) string
- func QueryStatus(endpoint string) (*transport.StatusResp, error)
- func ReadyFDFromEnv() *os.File
- func Spawn(cfg SpawnConfig) (pid int, err error)
- func Stop(cfg StopConfig) error
- type BusEntry
- type BusEntryState
- type DiscoverConfig
- type EntryStatus
- type SpawnConfig
- type SpawnFunc
- type StopConfig
Constants ¶
const DefaultStatusRPCTimeout = 2 * time.Second
DefaultStatusRPCTimeout caps how long QueryStatus waits for the bus to reply. 2s is generous — the bus's status_resp is a synchronous in-memory snapshot, sub-millisecond in practice; the timeout exists only to bound pathological cases (bus stuck in shutdown).
const DefaultStopTimeout = 5 * time.Second
DefaultStopTimeout is the wall-clock budget Stop waits for the bus to exit after the signal is sent. 5s covers the bus's own graceful tear-down (broadcast Bye → consumer goroutines drain → cleanup) with margin.
const ReadyFDEnv = "DWS_EVENT_BUS_READY_FD"
ReadyFDEnv is the env var the spawned `event _bus` child inspects to find the ready-pipe write end. The parent passes the FD number; child opens it via os.NewFile(fd, "ready") and writes 'R' on success or 'E' on failure. 3 is the first FD slot beyond stdio in cmd.ExtraFiles.
Variables ¶
var ErrNotRunning = errors.New("busctl: bus is not running")
ErrNotRunning indicates bus.lock either does not exist or its recorded PID is not alive. Stop returns this as a sentinel so the caller can distinguish "nothing to stop" from "failed to stop".
var ErrSpawnFailed = errors.New("busctl: bus child reported startup failure on ready pipe")
ErrSpawnFailed is returned when the child reports startup failure via the ready pipe ('E' byte). The child's exit error / log file holds the actual cause; this sentinel just lets the caller distinguish "ready pipe said no" from "ready pipe timed out / closed early".
var ErrSpawnTimeout = errors.New("busctl: bus child did not signal readiness within deadline")
ErrSpawnTimeout is returned when ReadyTimeout elapses without any signal.
var ReadyTimeout = 10 * time.Second
ReadyTimeout caps how long Spawn waits for the child to signal readiness. 10s is generous — bus startup is local-only work (file I/O + socket bind), so 1s would normally suffice; the extra headroom covers cold-start keychain prompts and slow CI machines.
Functions ¶
func Discover ¶
func Discover(cfg DiscoverConfig) (net.Conn, error)
Discover returns a connected net.Conn to the bus for cfg.ClientID. If the bus is not running, Discover forks a new one and waits for it to come up.
Race-free three-step (plan §12 P3):
- try dial IPC endpoint → success → done
- failed → call Spawn (fork _bus); Spawn blocks until ready pipe says 'R' (or returns ErrSpawnFailed if another process won the race and our bus startup hit ErrBusy on the lock)
- dial again — should succeed; retry with backoff up to DialDeadline in case Spawn succeeded but socket bind has tiny latency
On concurrent Discover by N processes: only one Spawn succeeds (the others get ErrBusy via the bus daemon's own lock acquisition). Losers fall through to the retry-dial loop in step 3 and connect to the bus the winner brought up.
func LockPath ¶
LockPath returns the canonical bus.lock path for the given working dir. Centralised so consume / status / stop all agree on the location.
func QueryStatus ¶
func QueryStatus(endpoint string) (*transport.StatusResp, error)
QueryStatus dials the bus IPC, sends Hello with Role=status, sends a StatusReq, reads exactly one StatusResp, and closes. Returns the decoded response or an error if any step fails.
Used by `dws event status` and `dws event list` to fetch live per-consumer / per-event-type counters. The bus's handleStatusRPC path (see internal/event/bus/daemon.go) handles this connection without registering with the Hub — ad-hoc tooling does not count as a consumer in `status.active_consumers`.
func ReadyFDFromEnv ¶
ReadyFDFromEnv returns the inherited ready pipe (or nil if not set). The `event _bus` command handler calls this at startup, passes the returned *os.File to bus.Run as Config.ReadyPipe, and the bus signals readiness through it.
func Spawn ¶
func Spawn(cfg SpawnConfig) (pid int, err error)
Spawn forks a detached `dws event _bus --client-id <id>` child process and waits for it to signal readiness via the ready pipe. Returns the child's PID on success — the caller can then dial the bus IPC endpoint.
stdio detach (plan invariant #7):
- cmd.Stdout / cmd.Stderr set to nil so the child's own writes don't pollute the parent's NDJSON stream
- Setsid on Unix so the child survives parent SIGHUP / parent exit
- CREATE_NEW_PROCESS_GROUP on Windows (set in spawn_windows.go)
Child startup (handled by the eventcmd._bus handler, P6):
- Opens os.NewFile(<DWS_EVENT_BUS_READY_FD>, "ready")
- On startup success → writes 'R' and closes
- On startup failure → writes 'E' and closes (child exits)
Parent (this function):
- Holds the read end open until either 1 byte is read or ReadyTimeout
- Returns ErrSpawnFailed for 'E', ErrSpawnTimeout otherwise
func Stop ¶
func Stop(cfg StopConfig) error
Stop signals the bus daemon for cfg.WorkDir to exit gracefully and waits for the process to actually die. Returns ErrNotRunning if no bus is running for that work dir.
Implementation note: on Unix we send SIGTERM. The bus daemon's Run loop watches its parent ctx for cancellation; the cobra `event _bus` command wires signal.NotifyContext so SIGTERM triggers ctx.Done() → graceful shutdown path. On Windows we use os.Process.Signal(os.Interrupt) which the Go runtime maps to TerminateProcess for processes outside our console group; for v1 that's acceptable (Windows graceful shutdown is future work — plan §16 v2).
Types ¶
type BusEntry ¶
type BusEntry struct {
WorkDir string `json:"workdir"`
Edition string `json:"edition"`
SourceKind dwsevent.SourceKind `json:"source_kind,omitempty"`
ClientIDHash string `json:"client_id_hash"`
IdentityHash string `json:"identity_hash,omitempty"`
HolderPID int `json:"holder_pid"`
State BusEntryState `json:"state"`
// Meta, if non-nil, lets list/status display the original ClientID
// (reverse-mapped from the hash) and the bus start time.
Meta *bus.Meta `json:"meta,omitempty"`
}
BusEntry is one bus working directory found on disk plus its detected lifecycle state. EnumerateBuses produces these; the cobra layer joins them with QueryStatus output to render the full status view.
func EnumerateBuses ¶
EnumerateBuses scans <configDir>/events/<editionFilter>/*/ for bus working directories. An empty editionFilter scans every edition directory found under events/.
Returns a deterministic slice sorted by (edition, source_kind, identity_hash). Missing/inaccessible directories are skipped silently — list/status commands should still succeed when only some editions have ever run a bus.
func FindBusByClientID ¶
FindBusByClientID is the "current ClientID" lookup used by `event status` (no --all). Returns the entry for the given (edition, clientIDHash) pair or nil if no bus has ever started for it. The caller derives the hash using event.ClientIDHash.
func FindBusByIdentity ¶
func FindBusByIdentity(configDir, editionName string, sourceKind dwsevent.SourceKind, identityHash string) *BusEntry
FindBusByIdentity looks up a bus in the source-kind-aware layout.
func (BusEntry) IPCEndpoint ¶
IPCEndpoint returns the IPC endpoint for this entry. Delegates to dwsevent.IPCEndpoint so status/stop dial exactly where consume and the bus daemon bound (including the short-path fallback when WorkDir is too deep for sun_path).
type BusEntryState ¶
type BusEntryState string
BusEntryState classifies a discovered bus directory's runtime state. Used by `dws event status/list` to render the table and by --fail-on-orphan to drive exit code.
const ( // BusStateRunning: bus.lock holds an alive PID — the daemon is up. BusStateRunning BusEntryState = "running" // BusStateOrphan: bus.meta exists but bus.lock PID is dead. The user // should `dws event stop --client-id <id>` (which detects the dead // PID and unblocks fresh starts) or rm -rf the working directory. BusStateOrphan BusEntryState = "orphan" // BusStateNotRunning: directory exists (e.g. bus.meta retained for // historic reasons) but bus.lock is missing or empty. Clean state. BusStateNotRunning BusEntryState = "not_running" )
type DiscoverConfig ¶
type DiscoverConfig struct {
WorkDir string
IPCEndpoint string
ClientID string
// Spawn is the fork-bus callback. Default busctl.Spawn.
Spawn SpawnFunc
// SpawnExtraArgs is forwarded to Spawn (for tests).
SpawnExtraArgs []string
// DialBackoff: initial sleep between retry dials when another process
// is spawning. Doubled each attempt up to DialMaxBackoff.
DialBackoff time.Duration
// DialMaxBackoff caps backoff.
DialMaxBackoff time.Duration
// DialDeadline caps total wall-clock time spent discovering.
DialDeadline time.Duration
}
DiscoverConfig describes one discover attempt. WorkDir holds bus.lock and usually (on Unix) bus.sock — see dwsevent.IPCEndpoint for the short-path fallback when WorkDir is too deep; the caller must mkdir it with pkg/config.DirPerm beforehand.
type EntryStatus ¶
type EntryStatus struct {
Entry BusEntry `json:"entry"`
Live *transport.StatusResp `json:"live,omitempty"`
}
EntryStatus combines static FS info (BusEntry) with the live RPC snapshot (StatusResp). For not_running / orphan entries Live is nil.
func QueryEntry ¶
func QueryEntry(entry BusEntry) EntryStatus
QueryEntry fetches the live status for one BusEntry. Returns the entry wrapped with a nil Live when state != running (or when the dial fails). Errors from QueryStatus are folded into Live=nil so the caller's table rendering does not need to surface per-bus dial failures (they are already conveyed by State).
type SpawnConfig ¶
type SpawnConfig struct {
// ExecPath is the dws binary to exec. Default os.Executable().
ExecPath string
// ClientID is passed as `--client-id` to `dws event _bus`. Required.
ClientID string
// ExtraArgs are appended after `--client-id`. Empty for normal use; tests
// pass `--extra-flag-for-test` etc.
ExtraArgs []string
// Env to pass to the child. Defaults to os.Environ(). The ReadyFDEnv
// entry is appended automatically.
Env []string
}
SpawnConfig describes one spawn attempt. ClientID is the only field inspected by the child; the rest govern process attributes the parent applies before exec.
type SpawnFunc ¶
type SpawnFunc func(SpawnConfig) (pid int, err error)
SpawnFunc abstracts the fork-bus operation so tests can inject a fake instead of execing a real binary. Production callers pass busctl.Spawn.
type StopConfig ¶
type StopConfig struct {
// WorkDir holds bus.lock; Stop reads the PID from there.
WorkDir string
// Timeout is the total wall-clock budget for graceful exit. After this,
// Stop returns an error; it does NOT escalate to SIGKILL — leave that
// to the operator.
Timeout time.Duration
}
StopConfig identifies the target bus and tunes timing.
Source Files