peer

package
v0.116.3 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 9, 2026 License: MIT Imports: 22 Imported by: 0

Documentation

Overview

Package peer manages this kojo binary's peer identity in the multi- device storage cluster (docs/multi-device-storage.md §3.5–§3.7).

At startup the binary loads (or, on first ever run, generates) a stable {device_id, name} pair that survives across restarts and binary upgrades. With docs/peer-simplify-plan.md step 9, the Ed25519 keypair that used to live here was retired — peer identity is now anchored in the Bearer tokens delivered through the pairing flow and in the network-layer cert (Tailscale TLS or operator-supplied). The pair survives only as the join-request payload + the operator-facing audit row.

Storage (all in kv namespace="peer", scope=machine — never crosses a peer boundary):

  • peer/device_id : string, RFC 4122 UUID v4
  • peer/name : string, os.Hostname() at first run

Older deploys also carried peer/public_key + peer/private_key; migration 0010 deletes those rows so a fresh install never sees them and an upgrade installs cleans them out alongside the peer_registry column drop.

Index

Constants

View Source 
const (
	StatusOpUpsert = "upsert"
	StatusOpTouch  = "touch"
	StatusOpExpire = "expire"
	// StatusOpDelete signals the row was removed from peer_registry
	// (operator DELETE). Unlike expire — which flips status to
	// offline but keeps the row — delete means the peer is gone
	// entirely, so a Subscriber drops it from its live cache rather
	// than storing an offline entry that would linger forever.
	StatusOpDelete = "delete"
)

StatusEventOp enumerates the Op string values the publisher uses. Subscribers compare against these constants rather than the bare strings so a typo on either side surfaces at compile time.

View Source 
const (
	KVNamespace = "peer"
	KeyDeviceID = "device_id"
	KeyName     = "name"
)

KV namespace + key constants. Single source of truth — exported so the migrate / clean / debug tooling can address the rows by name without re-deriving them.

View Source 
const AgentLockLeaseDuration = 5 * time.Minute

AgentLockLeaseDuration is how long each Acquire / Refresh grants the lock for. The refresh loop runs at half this cadence so a single missed refresh still leaves headroom before lease expiry. 5 minutes is enough that brief DB hiccups don't surrender locks without holding a stale claim too long after a crash: in the worst-case (binary crashes silently without Stop), another peer has to wait up to the full lease (5min) before it can steal, which is longer than the OfflineSweeper's 150 s (5×HeartbeatInterval) "mark offline" threshold but is intentional: the offline mark only updates registry liveness; lock ownership has stronger correctness needs and warrants a longer holdoff so a flaky network doesn't churn locks across peers.

View Source 
const AgentLockRefreshInterval = AgentLockLeaseDuration / 2

AgentLockRefreshInterval is how often the guard's refresh loop runs Refresh on every held lock. Half the lease so a single missed refresh still has lease remaining.

View Source 
const HeartbeatInterval = 30 * time.Second

HeartbeatInterval is how often Registrar.heartbeatLoop calls TouchPeer to refresh last_seen.

View Source 
const HubRowRefreshInterval = 5 * time.Minute

HubRowRefreshInterval is how often Discovery re-fetches hub-info AFTER approval to detect a NodeKey change on the Hub side (e.g. Hub upgrade that swaps the advertised NodeKey, Tailscale key rotation). Without this refresh the local peer_registry.Hub-row would freeze on the NodeKey observed at first pairing and any later Hub→peer call signed by a different NodeKey would 403 forbidden — exactly the failure mode the §3.7 inter-peer surface hit when Hub switched its advertised NodeKey from tsnet's to the host tailscaled's.

View Source 
const JoinHeartbeat = 60 * time.Second

JoinHeartbeat is the polling cadence while a join-request sits in `pending`.

View Source 
const OfflineThreshold = 5 * HeartbeatInterval

OfflineThreshold is the silence window after which a peer's `peer_registry.status` is flipped from 'online' to 'offline' by the sweeper. The convention from docs/multi-device-storage.md §3.10 is "5 missed heartbeats = expire". 5 × HeartbeatInterval (30s default) = 150s.

View Source 
const PairingProtocolVersion = 2

PairingProtocolVersion is the wire version of the peer pairing protocol (hub-info + join-request). Bumped whenever the auth / identity contract changes in a way that makes a registry row minted under the prior version unsafe to honour.

History: 

v1 — Bearer-token issuance + Ed25519 public_key in peer_registry.
     Retired in commit 0b6a5ff (NodeKey-only auth).
v2 — Tailscale NodeKey-only. Identity is bound exclusively via
     tsnet WhoIs; no shared secrets cross the wire.

Bumping this constant MUST be paired with a migration that wipes peer_registry / peer_pending so peers that previously paired under the old version cannot accidentally re-attach under the new auth contract. See migration 0017_pairing_protocol_v2.sql.

View Source 
const PeerNameMaxBytes = 255

PeerNameMaxBytes is the upper bound on peer_registry.name. 255 is the typical hostname limit; the FQDN + port form `<host>.<tailnet>.ts.net:8080` fits comfortably.

View Source 
const SweepInterval = HeartbeatInterval

SweepInterval is how often the OfflineSweeper checks for stale peers. It runs at the heartbeat cadence so a freshly-stale peer is detected within one heartbeat tick of crossing the threshold; any faster would hammer the DB without value (heartbeats only update last_seen at HeartbeatInterval anyway), any slower would let "online" rows linger past the documented threshold.

View Source 
const UnsafeNodeKey = "unsafe:local"

UnsafeNodeKey is the sentinel NodeKey the unsafe-mode middleware stamps on every request. It is NOT a real Tailscale key — the `unsafe:` prefix makes it distinguishable from anything tsnet could ever return so a row carrying it can never collide with a legitimate WhoIs lookup.

Variables

View Source 
var ErrAmbiguousPeerName = errors.New("peer: ambiguous name; multiple peer_registry rows matched")

ErrAmbiguousPeerName is returned by ResolvePeerTarget when an operator-supplied name matches more than one peer_registry row. The caller surfaces this with a 400 listing the candidate device_ids so the operator can disambiguate.

View Source 
var ErrCorrupt = errors.New("peer: corrupt identity (mixed kv row presence)")

ErrCorrupt is returned when peer/* kv rows are in a mixed state (some present, some missing) — see LoadOrCreate's contract.

View Source 
var ErrNoNodeKey = errors.New("peer: no Tailscale NodeKey for remote address")

ErrNoNodeKey is returned by NodeKeyResolver implementations when the remote address could not be mapped to a Tailscale NodeKey (the typical cause: the connection arrived on a non-tsnet listener, or tailscaled does not know the peer).

Functions

func HostFromRegistryName 

func HostFromRegistryName(name string) string

HostFromRegistryName extracts the case-folded hostname portion of a peer_registry.name value, stripping any scheme prefix and any :port suffix. Used by NameMatches so a single Tailscale machine name typed by a human ("bravo") matches the registry row regardless of whether the operator stamped it as "bravo", "bravo:8080", "http://bravo:8080", "bravo.tailnet.ts.net", or "https://bravo.tailnet.ts.net:8443". Returns "" when the input is empty or unparseable; callers treat that as "no match".

func IsDialAddress 

func IsDialAddress(name string) bool

IsDialAddress is the strict shape gate the operator-facing surfaces (CLI `--peer-add`, HTTP /api/v1/peers POST/PATCH, GUI register form) use to decide whether the supplied URL is a dial-form other peers can actually reach. The runtime stamps exactly two shapes:

  • Hub (tsnet): "<fqdn>:<port>" — no scheme; the historical Tailscale TLS form. peerSubscriberTargetsLoop's NormalizeAddress fills in "https://" for these.
  • Peer (`--peer`): "http://<ts-ipv4-or-host>:<port>" — explicit scheme so the Subscriber dials plain HTTP.

Anything else — bare hostname (`TVT-DEV-0000`), path/query/ fragment, unsupported scheme — would silently land in a registry row and produce an unreachable target the Subscriber / proxy loops over forever. Refuse so the bug surfaces at registration time.

`https://host[:port]` is allowed too for an operator who hand- stamps a non-default explicit scheme into the row; we just won't generate it ourselves.

func NameMatches 

func NameMatches(rowName, input string) bool

NameMatches returns true when input (a user-supplied identifier like "bravo" or "bravo.tailnet.ts.net") matches the given peer_registry.name. Match rules:

  • case-insensitive
  • input must equal the registry name's hostname (port + scheme stripped), OR
  • input must equal the leftmost DNS label of the registry name's hostname (so "bravo" matches a row whose host is "bravo.tailnet.ts.net").

IP literal rows (v4 or bracketed v6) match only on exact host equality — there's no "leftmost label" to peel.

func NoKeepAliveHTTPClient 

func NoKeepAliveHTTPClient(timeout time.Duration) *http.Client

NoKeepAliveHTTPClient returns an *http.Client with the same no-keep-alive transport NewPullClient uses internally, configured with the caller's timeout. This client predates PairingProtocolVersion v2: it was meant for every peer-signed outbound request, whose Ed25519/Bearer Authorization header carried a single-use nonce that a stale-conn retry could replay. Current callers (PullClient, Subscriber) send no Authorization header at all — identity travels via tsnet WhoIs — but they still use this client for its no-keep-alive transport.

func NormalizeAddress 

func NormalizeAddress(rawName string) (string, error)

NormalizeAddress turns a peer_registry.name value into a base URL that can be dialed (Subscriber WS, blob-pull GET, switch-device POST, ...). Accepted forms:

  • "host:port" (no scheme) — historical Hub-side row stamped by refreshPublicNameFromTailscale; resolved to "https://host:port" because tsnet listens on TLS.
  • "http://host:port" / "https://host:port" — daemon-only peers (`kojo --peer`) emit the http form; the Hub may also write explicit https for FQDN aliases. Case-insensitive on the scheme so url.Parse's RFC-3986 lowering doesn't trip us.

Anything else (other scheme, missing host, unparseable) returns a non-nil error; callers typically log and skip the row.

The returned URL has no path / query / fragment — only `scheme://host:port`. Callers append their own path segment.

func ResolvePeerTarget 

func ResolvePeerTarget(ctx context.Context, st PeerLookupStore, idOrName string) (*store.PeerRecord, error)

ResolvePeerTarget accepts either a canonical UUID device_id or a Tailscale machine name and returns the matching peer_registry row. UUID input goes straight to GetPeer; name input is matched case-insensitively against every row via NameMatches (full host or leftmost label).

Returns store.ErrNotFound when no match exists. Returns ErrAmbiguousPeerName (wrapped with the candidate IDs) when multiple rows match — the caller MUST disambiguate by device_id rather than guess.

Self-rows are intentionally NOT filtered here: the caller (orchestrator) refuses target==self with its own message and a 400, so blanking the row in resolution would muddy the diagnostic.

func ValidateDeviceID 

func ValidateDeviceID(id string) error

ValidateDeviceID enforces the canonical 8-4-4-4-12 lowercase UUID form. uuid.Parse on its own is too permissive (accepts URN form, braced form, raw bytes, uppercase) — letting any of those through would let the same logical UUID land twice in peer_registry under different keys and bypass self-detection by submitting an alternate spelling of the local device id. Empty input is a distinct error so callers can produce clearer messages ("required" vs "invalid format").

Shared by the HTTP handler (internal/server/peer_handlers.go) and the CLI (cmd/kojo/peer_cmd.go) so the same shape gate applies regardless of entry point.

func ValidateName 

func ValidateName(name string) error

ValidateName checks the human-readable peer name. Trimmed length > 0 and ≤ PeerNameMaxBytes; all Unicode control characters rejected so a UI rendering the value can't be tricked into ANSI escape / null / DEL / TAB injection. unicode.IsControl covers the C0 range (NUL, TAB, LF, CR, ESC), DEL (U+007F), and the C1 range (U+0080..U+009F).

Types

type AgentLockGuard 

type AgentLockGuard struct {
	// contains filtered or unexported fields
}

AgentLockGuard maintains live lock rows for a set of agent IDs.

Concurrency: the guard owns one goroutine for the refresh loop. AddAgent / RemoveAgent are safe to call from other goroutines; they take the in-memory mutex and either schedule the next Acquire or stop refreshing the dropped id.

State:

  • `desired`: the set of agents we WANT a lock on. AddAgent extends it; RemoveAgent / explicit shutdown shrinks it.
  • `tokens`: subset of `desired` whose Acquire has succeeded and whose row we are currently refreshing. Entries here come and go as the loop notices loss/regains the lock.

Each refresh tick first refreshes everything in `tokens`, then retries Acquire on the difference (desired \ tokens). That covers the "first Acquire returned ErrLockHeld" case (another peer was holding it; we re-poll until their lease expires and we steal) and the "refresh failed and the row vanished" case (next tick reseeds).

func NewAgentLockGuard 

func NewAgentLockGuard(st *store.Store, id *Identity, logger *slog.Logger) *AgentLockGuard

NewAgentLockGuard wires the deps. The caller is responsible for invoking Start once and Stop at shutdown.

func (*AgentLockGuard) AddAgent 

func (g *AgentLockGuard) AddAgent(ctx context.Context, agentID string)

AddAgent acquires the lock for an agent that was created after Start. Idempotent — calling on an already-held id is a no-op.

func (*AgentLockGuard) RemoveAgent 

func (g *AgentLockGuard) RemoveAgent(ctx context.Context, agentID string)

RemoveAgent releases the lock for an agent that was deleted or archived. Idempotent.

func (*AgentLockGuard) Start 

func (g *AgentLockGuard) Start(ctx context.Context, agentIDs []string)

Start launches the refresh goroutine and acquires the initial batch of locks listed in `agentIDs`. Individual Acquire failures are logged and skipped — the binary keeps running without a lock for the affected agent (matches the "best-effort" v1 stance).

AcquireFail-class errors:

  • ErrLockHeld: another live peer holds the lock. Logged Warn; the agent runs locally without owning the row, and the next refresh tick will retry (in case the other peer's lease expires).
  • any other store error: logged Error.

func (*AgentLockGuard) Stop 

func (g *AgentLockGuard) Stop()

Stop signals the refresh loop to exit and releases every held lock via ReleaseAgentLockByPeer. Bounded so shutdown can't block. Idempotent.

type Discovery 

type Discovery struct {
	// contains filtered or unexported fields
}

Discovery is the long-running auto-pairing coordinator.

func NewDiscovery 

func NewDiscovery(cfg DiscoveryConfig, identity *Identity, st *store.Store, logger *slog.Logger) (*Discovery, error)

NewDiscovery wires a Discovery.

func (*Discovery) Run 

func (d *Discovery) Run(ctx context.Context)

Run blocks until ctx is cancelled. The initial pass loops until the join is approved AND the Hub row in peer_registry carries a non-empty node_key; once that landed, it hands off to refreshLoop which keeps re-fetching hub-info and re-stamping the row on NodeKey drift. Lifecycle exits only on ctx cancellation. A mid-life pairing-protocol drift surfaces by bouncing back here (refreshLoop returns) so the outer for re-enters the join dance under the new contract.

The "Hub NodeKey landed" initial-pass condition closes a critical race flagged in the Codex re-review:

  • peer hits Hub immediately after Hub boot.
  • Hub's tsnet hasn't finished its login handshake yet → hub-info returns NodeKey="" .
  • peer stamps an empty node_key into its local peer_registry.
  • Owner approves; discovery exits.
  • Later, Hub dials this peer (Subscriber WS, blob push). The peer's tsnet middleware resolves Hub's NodeKey via WhoIs and looks it up in peer_registry — finds the row with an EMPTY node_key. Mismatch → caller stays Guest → 403 → ghosted peer.

Fix: discovery does not hand off to refreshLoop until peer_registry carries the Hub's real NodeKey. We accept the latest Hub spec returned in JoinResponse on every poll and rewrite the row.

func (*Discovery) SetPeerPublicURL 

func (d *Discovery) SetPeerPublicURL(u string)

SetPeerPublicURL lets main.go update the advertised URL once the listener has bound.

type DiscoveryConfig 

type DiscoveryConfig struct {
	HubURLOverride string
	DefaultHubPort int
	PeerPublicURL  string
}

DiscoveryConfig parameterises NewDiscovery.

type EventBus 

type EventBus struct {
	// contains filtered or unexported fields
}

EventBus is the in-memory pub/sub for peer_registry status mutations. Publishers (the Registrar's heartbeat loop, the OfflineSweeper, the HTTP handlers that UpsertPeer / DeletePeer) call Publish; subscribers (the cross-peer WS handler, the in-process peer subscriber) call Subscribe and read from the returned channel.

Why an in-memory bus instead of the existing internal/eventbus (which fans out events table rows): peer_registry mutations aren't currently routed through RecordEvent, and adding events table entries for every heartbeat would add a row every 30 s per peer — significant noise on the events stream that the /api/v1/changes cursor walks. A dedicated channel keeps the peer-status feed isolated from the domain-events feed.

Concurrency: Publish is non-blocking — if a subscriber's channel is full, its subscription is cancelled immediately (see Publish) rather than silently dropping the event. The subscriber's next read observes the closed channel and re-fetches the full peer_registry. This trades best-effort delivery for guaranteed publisher progress — a stalled HTTP client can't pin the heartbeat loop.

func NewEventBus 

func NewEventBus() *EventBus

NewEventBus returns a fresh bus with no subscribers.

func (*EventBus) Publish 

func (b *EventBus) Publish(evt StatusEvent)

Publish fans out evt to every live subscriber. Non-blocking: a subscriber whose channel is full has its subscription cancelled — the WS handler observes the channel close and emits a PolicyViolation close code that the peer's subscriber reconnects through (and re-receives the snapshot frame). This trades a noisy log line for guaranteed publisher progress AND correctness: previously the publisher bumped a counter and dropped the event silently, leaving the subscriber with a stale view it had no way to detect.

func (*EventBus) Subscribe 

func (b *EventBus) Subscribe(ctx context.Context) (<-chan StatusEvent, func(), error)

Subscribe returns a channel that will receive every StatusEvent the bus publishes from now on. The channel has a small buffer (subBufferSize) — a subscriber that doesn't drain it within that many events starts dropping. The returned cancel function removes the subscription and closes the channel; subscribers MUST call it on exit to avoid leaks.

The parent ctx also tears down the subscription: when it cancels, the bus's internal context derived from it fires the cancel goroutine that removes the subscriber.

func (*EventBus) Subscribers 

func (b *EventBus) Subscribers() int

Subscribers returns the current subscriber count. Exposed for tests + telemetry; production code shouldn't care.

type HubInfo 

type HubInfo struct {
	DeviceID        string `json:"deviceId"`
	Name            string `json:"name"`
	URL             string `json:"url"`
	NodeKey         string `json:"nodeKey,omitempty"`
	Version         string `json:"version"`
	ProtocolVersion int    `json:"protocolVersion"`
}

HubInfo is the response shape of GET /api/v1/peers/hub-info.

NodeKey is the Hub's Tailscale stable NodeKey. The peer stamps it onto its local peer_registry row for the Hub so the tsnet identity middleware can later resolve inbound Hub requests (Subscriber WS, blob push, agent-sync) to RolePeer. Empty until the Hub's tsnet has finished its login handshake; the peer re-fetches hub-info on the next discovery tick in that case.

ProtocolVersion is the pairing protocol the Hub advertises (see PairingProtocolVersion). Enforced by the discovery loop: a mismatch causes the peer to wipe its local peer_registry row for this Hub and retry on the next tick (see Run). The Hub also re-validates the field on /join-request, so neither end can keep a half-paired record across a version boundary.

type Identity 

type Identity struct {
	DeviceID string
	Name     string
}

Identity is the loaded-or-generated peer identity. After docs/peer-simplify-plan.md step 9 it carries only the device_id + human-readable name; auth material lives in peer_tokens (server side) and in the OutBearerNS kv (client side).

func LoadOrCreate 

func LoadOrCreate(ctx context.Context, st *store.Store) (*Identity, error)

LoadOrCreate returns the persisted peer identity. On first run it generates a UUID v4 + the hostname-derived name and persists both kv rows; subsequent runs read them back.

Error contract:

  • Partial-row state (one present, the other missing) returns ErrCorrupt; recovery requires either restoring kojo.db from a snapshot or deleting the surviving row so LoadOrCreate can mint a fresh identity. Auto-recovery is intentionally skipped so a half-rotated identity surfaces visibly.
  • kv I/O errors propagate verbatim.

type JoinResponse 

type JoinResponse struct {
	State string   `json:"state"`
	Hub   *HubInfo `json:"hub,omitempty"`
}

JoinResponse is the response shape of POST/GET /api/v1/peers/join-request.

type NodeKeyResolver 

type NodeKeyResolver func(ctx context.Context, remoteAddr string) (string, error)

NodeKeyResolver maps an incoming HTTP request's remote address to the Tailscale stable NodeKey of the caller. The Server-side identity middleware uses this to find the matching peer_registry row.

remoteAddr is the value of *http.Request.RemoteAddr (`ip:port`). Implementations forward to tsnet.LocalClient.WhoIs and return the String() form of the resolved tailcfg.Node.Key (the canonical `nodekey:...` representation).

A nil resolver is treated as "no tsnet bound on this listener" — the middleware refuses every request that did not come in via a trusted path (e.g. --local + --unsafe).

type OfflineSweeper 

type OfflineSweeper struct {
	// contains filtered or unexported fields
}

OfflineSweeper is the v1 stand-in for the inter-peer WS subscriber described in docs/multi-device-storage.md §3.10. Until peers cross-subscribe over mTLS-secured WS connections (a follow-up slice that requires a peer-trust bootstrap path that v1 doesn't yet have), the Hub leans on heartbeat freshness alone:

  • Each peer's Registrar refreshes its self-row's last_seen on a heartbeat tick.
  • On the Hub, this sweeper periodically scans peer_registry and flips rows whose last_seen is older than OfflineThreshold from 'online' to 'offline'.

That gives the cluster eventual consistency on liveness without needing peer-to-peer connectivity beyond their connection to the Hub. Cross-WS subscription would only narrow the detection window from "<=2× HeartbeatInterval" to "near-realtime", which v1 does not need.

The sweeper excludes its own peer's row so a stuck sweeper goroutine never flips the local self-row offline while the registrar's heartbeat is still firing.

Concurrency: identical model to Registrar — one goroutine, sync.Once shutdown guard, sync.WaitGroup for graceful drain.

func NewOfflineSweeper 

func NewOfflineSweeper(st *store.Store, id *Identity, logger *slog.Logger) *OfflineSweeper

NewOfflineSweeper wires the deps. Returns nil-safe sentinels so a caller can always Start/Stop without nil-checks.

func (*OfflineSweeper) SetEventBus 

func (s *OfflineSweeper) SetEventBus(bus *EventBus)

SetEventBus wires the optional cross-peer status push channel. When non-nil, sweepOnce republishes one "expire" StatusEvent per flipped row so WS subscribers learn of the offline transition without waiting for their own heartbeat. Safe to call once before Start.

func (*OfflineSweeper) SetPresence 

func (s *OfflineSweeper) SetPresence(p *Presence)

SetPresence wires the in-memory active-connection set the events WS handler populates. Each sweep tick refreshes last_seen for every active deviceID before running the stale-sweep predicate, so a peer with a live WS is held online even on a flaky uplink where a missed touch tick would otherwise let the row age out. Safe to call once before Start.

func (*OfflineSweeper) Start 

func (s *OfflineSweeper) Start()

Start launches the sweep goroutine. Returns nil even if the deps are nil — the goroutine exits immediately on its first tick — so the caller can wire it unconditionally and let the route-guard pattern decide whether the Server actually exposes peer endpoints.

func (*OfflineSweeper) Stop 

func (s *OfflineSweeper) Stop()

Stop signals the loop to exit and waits for it. Idempotent.

type PeerLookupStore 

type PeerLookupStore interface {
	GetPeer(ctx context.Context, deviceID string) (*store.PeerRecord, error)
	ListPeers(ctx context.Context, opts store.ListPeersOptions) ([]*store.PeerRecord, error)
}

PeerLookupStore is the minimal store contract ResolvePeerTarget needs. *store.Store satisfies it; tests can pass a fake.

type Presence 

type Presence struct {
	// contains filtered or unexported fields
}

Presence tracks which remote peers currently hold an active peer-events WebSocket against this daemon. The peer_registry last_seen / status columns reflect a 5-missed-heartbeat sampling view — fine in steady state, brittle on a flaky mobile uplink where a single missed 15s touch can push last_seen past the 150s OfflineThreshold and cause the OfflineSweeper to flip a row offline even though the WS connection is alive.

Presence is the second source of truth: if at least one WS for a deviceID is currently held by this process, the peer is observably reachable RIGHT NOW regardless of what the sampled last_seen says. OfflineSweeper consults Presence.IsActive and skips any peer with a live connection, so a sweep tick can never demote an actively-connected peer.

Concurrency: a multi-set semantics on top of a single mutex. AddConn returns a release closure so handlers can `defer` the removal without tracking a per-call key. The same deviceID may hold multiple simultaneous connections (mobile reconnect race, dual-stack DNS, retries) — the counter only drops to 0 when every connection has been released.

func NewPresence 

func NewPresence() *Presence

NewPresence returns an empty presence map ready for use.

func (*Presence) ActiveDeviceIDs 

func (p *Presence) ActiveDeviceIDs() []string

ActiveDeviceIDs returns a snapshot of every deviceID with at least one active connection. Intended for diagnostics / /api/v1/peers debug endpoints; the OfflineSweeper hot path uses IsActive instead to avoid allocating a slice per sweep tick.

func (*Presence) AddConn 

func (p *Presence) AddConn(deviceID string) func()

AddConn marks deviceID as actively connected. Returns a release closure the caller MUST defer to drop the count when the connection ends. Empty deviceID is a no-op so call sites don't have to special-case anonymous peers. The release closure is idempotent (sync.Once) so a caller can wire it both into a defer AND an explicit early-cleanup path without double-decrementing the count from a parallel goroutine.

func (*Presence) IsActive 

func (p *Presence) IsActive(deviceID string) bool

IsActive reports whether at least one connection is currently held for deviceID. Nil-safe: a nil Presence returns false so the OfflineSweeper degrades gracefully to last_seen-only when the binary wasn't wired with a presence set (e.g. unit tests).

type PullClient 

type PullClient struct {
	// contains filtered or unexported fields
}

PullClient drives outbound GET /api/v1/peers/blobs/* requests against a single source peer. Reuse one *PullClient across many PullOne calls — the *http.Client is shared but its transport disables keep-alives, so each PullOne opens a fresh TCP/TLS connection. The single-use connection policy is what prevents Go's idempotent-GET stale-conn retry from re-sending the same signed nonce; see NewPullClient for the full rationale.

func NewPullClient 

func NewPullClient(id *Identity, httpClient *http.Client, logger *slog.Logger) *PullClient

NewPullClient wires the client. Pass nil for httpClient to use a default with a sane timeout; tests can inject a fixture client.

The default transport DISABLES connection keep-alive, forcing a fresh TCP/TLS handshake per request. Historically this guarded against Go's transport silently retrying an idempotent GET on a stale-connection error and replaying the signed nonce that the now-retired Ed25519/Bearer auth schemes stamped into the Authorization header (see version.go's PairingProtocolVersion history). PullOne sends no Authorization header today — identity travels via tsnet WhoIs on the receiving side — so that hazard no longer applies, but the no-keep-alive transport itself is unchanged. Cost: a few extra handshakes per switch — negligible against the blob payload sizes.

func (*PullClient) PullMany 

func (c *PullClient) PullMany(ctx context.Context, src PullSource, items []PullItem, dst *blob.Store) ([]PullResult, error)

PullMany sequentially fetches every item from src into dst. A fatal local error (context cancel, signing failure) aborts the batch and returns the partial result list with the triggering error. Per-URI HTTP / sha256 failures are recorded in the result list and the batch continues; the caller decides whether to call handoff/abort based on the populated statuses.

Ordering is preserved so a caller mapping result[i] back to items[i] for logging works without rebuilding a map.

func (*PullClient) PullOne 

func (c *PullClient) PullOne(ctx context.Context, src PullSource, item PullItem, dst *blob.Store) (PullResult, error)

PullOne fetches a single blob body from src and writes it to dst via blob.Store.Put. The URI is the canonical kojo:// form (matches blob_refs.uri verbatim). When item.ExpectedSHA256 is non-empty (the orchestrator path), the helper enforces it as BOTH the X-Kojo-Blob-SHA256 response header value AND the ExpectedSHA256 passed to blob.Store.Put — a compromised source can therefore only succeed by returning a body whose actual hash matches the orchestrator's pre-existing blob_refs row. When ExpectedSHA256 is empty (legacy / drill path), the helper falls back to "trust the response header" verification, which is strictly weaker.

Failure modes that DO NOT abort the batch:

  • HTTP non-200 (handler returns 409 not_in_handoff / wrong_home, 410 body_missing, 404 not_found, ...): recorded as Status="http_status".
  • SHA256 mismatch (body hashed to something other than the X-Kojo-Blob-SHA256 header) — recorded as Status="sha256_mismatch"; the on-disk file is NOT updated (blob.Store.Put aborts before rename).
  • Header / orchestrator digest disagreement — recorded as Status="sha256_mismatch" without touching disk.

Failure modes that DO return an error to the caller:

  • Local I/O when constructing/signing the request, dialing the source, or wiring the response stream.
  • Context cancellation.

The pull is idempotent w.r.t. blob.Store: writing the same body twice produces the same digest and leaves blob_refs unchanged.

type PullItem 

type PullItem struct {
	URI            string `json:"uri"`
	ExpectedSHA256 string `json:"expected_sha256,omitempty"`
}

PullItem is one entry in a pull batch — the URI to fetch plus the orchestrator-asserted sha256 the body must hash to. The digest comes from the Hub's blob_refs row (the orchestrator reads it before dispatching the pull) so target's trust is rooted in the signed authority that drove the switch, not in the unsigned response header the source peer echoes. Empty ExpectedSHA256 falls back to "header-only" verification, which is strictly weaker — orchestrator callers should always populate it.

type PullResult 

type PullResult struct {
	URI    string `json:"uri"`
	Status string `json:"status"` // "ok" | "error" | "sha256_mismatch" | "http_status"
	SHA256 string `json:"sha256,omitempty"`
	Size   int64  `json:"size,omitempty"`
	Error  string `json:"error,omitempty"`
}

PullResult is the per-URI outcome of a pull batch.

type PullSource 

type PullSource struct {
	// DeviceID is the logical source peer's identity. When RelayVia
	// is nil, the request dials this peer directly; otherwise
	// DeviceID rides in the `?relay_from=` query so the relayer
	// (typically the Hub) knows which third peer to forward to.
	DeviceID string
	// Address is the base URL of whoever this client should dial.
	// Direct mode: source's URL. Relay mode: relayer's URL (Hub).
	Address string
	// RelayVia, when non-nil, makes PullOne dial RelayVia.Address
	// with `?relay_from=<DeviceID>` appended instead of dialing the
	// source directly. No Authorization header travels either way —
	// identity is established via tsnet WhoIs on the receiving side
	// (PairingProtocolVersion v2) — so relay mode exists to route
	// around network reachability, not to swap credentials. The Hub
	// side (peer_blob_handler.go relayPeerBlob) strips the query and
	// re-issues the upstream GET to the real source.
	RelayVia *PullSource
}

PullSource identifies the peer to fetch from.

type PushClient 

type PushClient struct {
	// contains filtered or unexported fields
}

PushClient is reusable. The *http.Client is shared but the transport disables keep-alives so each PushOne opens a fresh TCP/TLS connection — same reasoning as PullClient.

func NewPushClient 

func NewPushClient(id *Identity, httpClient *http.Client, logger *slog.Logger) *PushClient

NewPushClient wires the client. Pass nil for httpClient to use a no-keep-alive default; tests can inject a fixture client.

func (*PushClient) PushOne 

func (c *PushClient) PushOne(
	ctx context.Context,
	dst PushTarget,
	scope blob.Scope,
	blobPath string,
	expectedSHA256 string,
	body io.Reader,
	size int64,
) error

PushOne uploads body to dst's ingest endpoint as (scope, path). expectedSHA256 must be the hex digest of the body (pre-computed during the local blob.Store.Put) and is sent as X-Kojo-Expected-SHA256 so the hub aborts pre-rename on any mismatch.

Returns nil on a 2xx response. Any non-2xx surface as an error with the response body snippet preserved for diagnostics.

type PushTarget 

type PushTarget struct {
	// DeviceID is the hub peer's identity. Validated as part of the
	// input contract in PushOne; the actual auth is via tsnet WhoIs
	// on the receiving side, not a signed audience claim.
	DeviceID string
	// Address is the base URL of the hub (e.g.
	// "https://hub.tail-net.ts.net:8080"). The ingest path is
	// appended; any path / query already on the URL is dropped.
	Address string
}

PushTarget identifies the hub a non-hub peer pushes to.

type Registrar 

type Registrar struct {
	// contains filtered or unexported fields
}

Registrar maintains this peer's row in peer_registry: registers it at startup, refreshes last_seen on a heartbeat tick, and marks the peer offline at shutdown.

Concurrency: one goroutine owns the heartbeat loop; Stop signals shutdown via stopCh and waits on wg. Multiple Stop calls are idempotent (sync.Once guard).

func NewRegistrar 

func NewRegistrar(st *store.Store, id *Identity, logger *slog.Logger) *Registrar

NewRegistrar wires the deps. The caller MUST call Start before Stop; calling Stop on a never-Started Registrar is a no-op.

func (*Registrar) ClearSelfNodeKey 

func (r *Registrar) ClearSelfNodeKey(ctx context.Context) error

ClearSelfNodeKey wipes the self-row's node_key column AND the in-memory cache. Used at Hub-mode boot to drop a stale value left by a previous binary that stamped tsnet.Server's NodeKey here — the post-clear value is then refilled with the host tailscaled NodeKey via cmd/kojo's captureOSSelfNodeKeyForRegistrar. Without this, hub-info would keep advertising the stale tsnet NodeKey (which peers stamp into their local Hub-row but never observe on the inbound side, because Hub's outbound HTTP runs through http.DefaultTransport not tsnet.Server.Dial), and the §3.7 inter-peer surface would keep 403-ing forbidden.

func (*Registrar) RefreshPublicName 

func (r *Registrar) RefreshPublicName(ctx context.Context) error

RefreshPublicName re-upserts the self-row with the current publicURL. Called by cmd/kojo AFTER tsnet has reported its assigned FQDN — Start runs before tsnet is up, so the initial upsert lands with an empty URL and this method fills it in. Idempotent; safe to call repeatedly.

Refuses to run after Stop has been called — without that gate, a slow retry goroutine could fire the refresh AFTER Stop emitted its final offline-touch and end up flipping the row back to online, contradicting the shutdown signal other peers already received.

func (*Registrar) SetEventBus 

func (r *Registrar) SetEventBus(bus *EventBus)

SetEventBus attaches the cross-peer status push bus. Optional; when non-nil, each register / heartbeat / shutdown publishes a StatusEvent so subscribers receive near-realtime status changes without waiting for the OfflineSweeper. Safe to call once before Start.

func (*Registrar) SetPublicURL 

func (r *Registrar) SetPublicURL(url string)

SetPublicURL stamps the dial address other peers reach this row on. Other peers' Subscriber dials `https://<row.url>` and expects DNS-resolvable. cmd/kojo passes the Tailscale FQDN + port (e.g. "bravo.<tailnet>.ts.net:8080") here once tsnet has reported its assigned name; --peer hosts pass "http://<ts-ipv4>:<port>". Safe to call once before Start; empty leaves the column blank until a later call lands.

func (*Registrar) SetSelfNodeKey 

func (r *Registrar) SetSelfNodeKey(nk string)

SetSelfNodeKey records the local Tailscale NodeKey to stamp on the self-row's node_key column. cmd/kojo wires this from tsnet.LocalClient.Status (Hub) or localTailscale.Status (--peer). Empty leaves the column unchanged.

func (*Registrar) Start 

func (r *Registrar) Start(ctx context.Context) error

Start performs the initial peer_registry upsert (online + last_seen = now) and launches the heartbeat goroutine. Returns the upsert error verbatim if the row write fails — callers SHOULD log and proceed (the binary still works, just won't appear in cross-peer listings until the next start).

func (*Registrar) Stop 

func (r *Registrar) Stop()

Stop signals the heartbeat loop to exit, waits for it, then best-effort-marks the peer offline so cross-peer listings reflect the shutdown.

Safe to call multiple times (sync.Once); safe to call from a signal handler (no blocking I/O on the calling goroutine beyond the bounded shutdownTouchTimeout TouchPeer).

type StatusEvent 

type StatusEvent struct {
	DeviceID string `json:"device_id"`
	Name     string `json:"name,omitempty"`
	URL      string `json:"url,omitempty"`
	Status   string `json:"status"`              // "online" | "offline"
	LastSeen int64  `json:"last_seen,omitempty"` // unix millis
	// Op is "upsert" for register / first-touch, "touch" for
	// heartbeat status refreshes, "expire" for sweeper-driven
	// stale flips, "delete" for operator-driven row removal. Lets a
	// subscriber gate on which mutations matter without re-deriving
	// from before/after rows.
	Op string `json:"op"`
}

StatusEvent is one peer_registry mutation visible to subscribers of the cross-peer status feed (docs §3.10 "両方向 heartbeat" narrowed to push semantics). The fields are intentionally a minimal subset of PeerRecord — subscribers re-fetch the full row via the registry API if they need capabilities or public_key.

type Subscriber 

type Subscriber struct {
	// contains filtered or unexported fields
}

Subscriber maintains one connect-and-reconnect loop per target peer. No Authorization header is sent on the WS upgrade; the remote peer authenticates the caller via tsnet WhoIs on the receiving side (PairingProtocolVersion v2).

func NewSubscriber 

func NewSubscriber(id *Identity, st *store.Store, logger *slog.Logger) *Subscriber

NewSubscriber wires the deps. The Subscriber deliberately never re-publishes remote events onto the local EventBus (see handleFrame) — A↔B mutual subscriptions would otherwise reflect the same status_event back and forth forever (no origin/hop field to suppress the loop). In-process consumers consult Subscriber.LiveStatus instead.

func (*Subscriber) LiveStatus 

func (s *Subscriber) LiveStatus(deviceID string) (StatusEvent, bool)

LiveStatus returns the most recent StatusEvent received for deviceID across any target's feed. When the same deviceID appears in multiple targets the freshest LastSeen wins. Returns (zero-value, false) when never observed.

func (*Subscriber) SetTargets 

func (s *Subscriber) SetTargets(targets []SubscriberTarget)

SetTargets reconciles the connect set: any target in `targets` that doesn't have a goroutine yet gets one started; any running goroutine whose target is no longer in `targets` is cancelled. Concurrent-safe. No-op after Stop has been called.

func (*Subscriber) Stop 

func (s *Subscriber) Stop()

Stop signals every per-target loop to exit and waits for them. Idempotent.

type SubscriberTarget 

type SubscriberTarget struct {
	DeviceID string
	Address  string
}

SubscriberTarget identifies one peer the Subscriber should connect to. Address is the base URL (e.g. "https://peer-b.tail-net.ts.net:8443") and DeviceID is the remote peer's identity, keyed against the peer_registry row the Hub knows about.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.