hubcluster

package
v0.0.6 Latest Latest
Warning

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

 Go to latest Published: Oct 16, 2026 License: MIT Imports: 28 Imported by: 0

Documentation

Overview

Package hubcluster lets several `cloop ui` processes serve one control plane (Task 20354).

A hub used to be a singleton, and pkg/hublease made it one on purpose: the database was shared, but everything a hub keeps beside it lived in one process — the agent WebSocket it holds, the run whose output it streams, the dashboard sockets it pushes to, the sweeps it runs over shared rows. A second process would diverge quietly, so it was refused.

This package replaces the refusal with membership. Every hub process joins the control plane as a member and heartbeats; the members share four things through the database they already share: 

membership   who is serving, how to reach them, and whether they are alive
             (hub_members). Liveness is a heartbeat TTL, plus a pid probe
             that answers at once for a member on the same machine and pid
             namespace — the common crash is a supervisor restarting a
             killed process within seconds, and that should not cost a TTL.

leadership   the pre-existing hub lease (hub_instances), now held by one
             member at a time for the work that must happen exactly once:
             retention, backups, auto-resume, the session janitor. Losing
             it demotes the member; it no longer shuts the process down.

events       a bus (hub_bus) for what one member produces and every
             member's clients must see — a line of harness output, a run
             starting, an executor going offline. Batched on write, polled
             by sequence number on read.

ownership    which member holds a thing that cannot be shared (hub_owners):
             an agent's socket, a dispatched run, an in-flight login. A
             request about it that reaches another member is forwarded to
             the owner over an authenticated peer channel, and a thing whose
             owner died is adopted by a member able to reach it.

The SQL is pkg/statedb's (migrations/0052_hub_cluster.sql); policy — who is alive, who may adopt, how a peer proves it is one — is here.

What a cluster requires

One database file that every member opens through the same kernel: members on one machine, or containers on one node sharing a volume. SQLite's WAL needs shared memory between its connections, so it does not work over a network filesystem, and the bus's sequence-number cursor relies on SQLite having exactly one writer at a time. A control plane spread over several machines needs a network database, which is not this package's concern — but nothing here assumes the members share a pid namespace, a process table or a loopback interface: every cross-member question is answered from the database or over the peer channel.

Index

Constants

View Source 
const (
	// DefaultHeartbeat is how often a member renews its row and re-reads the
	// member list.
	DefaultHeartbeat = 5 * time.Second
	// DefaultMemberTTL is how long a member survives without renewing. Four
	// missed beats: a slow fsync or a GC pause must not hand a live member's
	// runs to someone else.
	DefaultMemberTTL = 20 * time.Second
	// DefaultBusPoll is how often a member reads the bus. It is the latency a
	// dashboard attached to one member sees for an event another produced.
	DefaultBusPoll = 250 * time.Millisecond
	// DefaultBusFlush bounds how long a published event waits to be batched.
	DefaultBusFlush = 50 * time.Millisecond
	// DefaultBusRetention is how long an event stays on the bus. Members read
	// it within a poll interval; the rest of the window is for a member that
	// stalled — past it, the member learns it missed events and resyncs.
	DefaultBusRetention = 2 * time.Minute
	// DefaultCampaign is how often a follower checks whether leadership is
	// free.
	DefaultCampaign = 5 * time.Second
)

Defaults. Chosen so a crashed member on another machine is noticed within half a minute, and one on this machine within a heartbeat.

View Source 
const (
	HeaderPeer      = "X-Cloop-Peer"
	HeaderPeerTo    = "X-Cloop-Peer-To"
	HeaderPeerTime  = "X-Cloop-Peer-Time"
	HeaderPeerNonce = "X-Cloop-Peer-Nonce"
	HeaderPeerIP    = "X-Cloop-Peer-Client-Ip"
	HeaderPeerSig   = "X-Cloop-Peer-Signature"
	// HeaderPeerHop names the member that answered, on responses, so a
	// client and a test can see where a request was served.
	HeaderServedBy = "X-Cloop-Served-By"
)

Header names on a forwarded request.

View Source 
const (
	DefaultProbeInterval = 5 * time.Second
	DefaultProbeTimeout  = 5 * time.Second
)

Probe defaults. A stuck member is judged within UnresponsiveAfter — the member TTL — plus one probe, which is the window in which a crashed member on another machine is noticed too.

View Source 
const DefaultOrphanRunGrace = 2 * time.Minute

DefaultOrphanRunGrace is how long a run whose owner died keeps its project marked running while no member has adopted it. Long enough for the agent behind the run to reconnect to another member and be adopted there; short enough that a run nobody can reach does not wedge its project.

View Source 
const DefaultPeerHeaderTimeout = 6 * time.Minute

DefaultPeerHeaderTimeout is how long a forwarded request waits for the owner's response headers. It has to outlast the slowest request a member answers only once its work is done — a voice transcription or a project's `cloop init` run on an edge agent, each bounded at five minutes — or the caller is told the member could not be reached while the work goes on there. A member that stops answering altogether is not left to this bound: the probe judges it unresponsive within its window, and the requests waiting on it are failed then (see probe.go).

View Source 
const OwnerKindRun = "run"

OwnerKindRun is the ownership kind of a project's run. The key is the project directory as the hub registered it.

View Source 
const PingPath = "/api/internal/cluster/ping"

PingPath is the internal endpoint a member answers probes on. pkg/ui serves it behind the peer signature, like every endpoint under the internal prefix.

Variables

View Source 
var ErrLegacyHub = errors.New("hubcluster: a hub that cannot join a cluster controls this state")

ErrLegacyHub is returned by Join when the control plane is held by a hub that predates clustering: it holds the lease but is not a member, so it neither forwards nor reads the bus, and serving beside it would be exactly the divergence the lease used to prevent.

View Source 
var ErrMemberUnresponsive = errors.New("hubcluster: the member is not answering")

ErrMemberUnresponsive is the cause of a request failed because the member it waited on was judged unresponsive, and the error a request to such a member fails with without being sent.

View Source 
var ErrNoRoute = errors.New("hubcluster: member advertises no URL")

ErrNoRoute means the target member advertises no URL to reach it by.

View Source 
var ErrNotPeer = errors.New("hubcluster: not a peer request")

ErrNotPeer is returned by VerifyPeer for a request with no peer headers.

Functions

func IsPeerRequest 

func IsPeerRequest(r *http.Request) bool

IsPeerRequest reports whether r claims to come from a peer. It does not verify the claim; VerifyPeer does.

func LiveMemberRows 

func LiveMemberRows(dbPath string, now time.Time) ([]statedb.HubMemberRow, error)

LiveMemberRows reads the member table at dbPath and returns the rows of members that are serving now, judged the way members judge each other. For callers outside a cluster — the CLI, an exclusive hub deciding whether it may start — that have no Node of their own.

func RowAlive 

func RowAlive(r statedb.HubMemberRow, now time.Time) bool

RowAlive judges one member row from this process, the way members judge each other: a leave is final, a heartbeat older than DefaultMemberTTL is dead, and a fresh one is alive unless it names a process on this machine, in this pid namespace, that is gone.

func RunClaimLive 

func RunClaimLive(o Owner, now time.Time, grace time.Duration) bool

RunClaimLive reports whether o, a claim on a project's run held by a member other than the observer, means the run is — or may still be — executing.

A live owner is conclusive. A dead owner's claim fails closed for grace after the owner was last seen: the run may be adopted by the member its agent reconnects to, and acting on the project in that window would reset tasks under a run that is about to resume. A claim the owner took before it asked any executor for anything has nothing behind it once the owner is gone, and does not count; one whose dispatch reached an executor may have started a workload, and counts until the grace runs out (Task 20419) — the leader stops that workload in the meantime. Metadata that cannot be decoded is judged as a claim with a workload, the reading that fails closed.

func StripPeerHeaders 

func StripPeerHeaders(h http.Header)

StripPeerHeaders removes anything a client sent that claims to be from a peer. Called before signing, so a forwarded request cannot smuggle a claim of its own through, and by a server on requests that failed verification.

func WithPeerCall 

func WithPeerCall(ctx context.Context, pc PeerCall) context.Context

WithPeerCall records a verified peer call in ctx.

Types

type BusStats 

type BusStats struct {
	Published uint64 `json:"published"`
	Delivered uint64 `json:"delivered"`
	Dropped   uint64 `json:"dropped"`
	Gaps      uint64 `json:"gaps"`
	Cursor    int64  `json:"cursor"`
}

BusStats reports counters for status output and tests.

type CallError 

type CallError struct {
	Member  string
	Status  int
	Message string
}

CallError is a non-2xx answer from a peer.

func (*CallError) Error 

func (e *CallError) Error() string

type Client 

type Client struct {
	// IP is the client's address.
	IP string
	// Proto is "https" or "http", the scheme the browser used; empty takes
	// the forwarding connection's own.
	Proto string
	// Host is the host[:port] the browser addressed; empty takes r.Host.
	Host string
}

Client is what the forwarding member resolved about the browser behind a request: its address, and the scheme and host it addressed — which a proxy in front of the forwarding member may have reported in X-Forwarded-* that the owner never sees. The owner judges the request's origin from them (Task 20394), so they travel as X-Forwarded-Proto and X-Forwarded-Host on the signed hop; the owner believes them because the hop is signed.

type Event 

type Event struct {
	Seq     int64
	Origin  string
	Target  string
	Topic   string
	Key     string
	Payload json.RawMessage
	At      time.Time
}

Event is one message delivered from the bus.

func (Event) Decode 

func (e Event) Decode(v any) error

Decode unmarshals the payload into v.

type LegacyHubError 

type LegacyHubError struct {
	Holder statedb.HubLeaseRow
	Age    time.Duration
	TTL    time.Duration
}

LegacyHubError names the pre-cluster hub that holds the control plane.

func (*LegacyHubError) Error 

func (e *LegacyHubError) Error() string

func (*LegacyHubError) Unwrap 

func (e *LegacyHubError) Unwrap() error

Unwrap lets errors.Is(err, ErrLegacyHub) match.

type Member 

type Member struct {
	ID           string            `json:"id"`
	Hostname     string            `json:"hostname"`
	PID          int               `json:"pid"`
	BootID       string            `json:"-"`
	Address      string            `json:"address"`
	AdvertiseURL string            `json:"advertise_url"`
	Version      string            `json:"version"`
	Endpoints    map[string]string `json:"endpoints,omitempty"`
	StartedAt    time.Time         `json:"started_at"`
	HeartbeatAt  time.Time         `json:"heartbeat_at"`
	LeftAt       time.Time         `json:"left_at,omitzero"`
	// Alive is the verdict at the time the member list was read.
	Alive bool `json:"alive"`
	// Unresponsive reports that the member reading the list judged this one
	// unresponsive: it heartbeats, and has failed every probe since
	// UnresponsiveSince (Task 20419). Such a member is treated as down for
	// adoption and forwarding while Alive stays true.
	Unresponsive      bool      `json:"unresponsive,omitempty"`
	UnresponsiveSince time.Time `json:"unresponsive_since,omitzero"`
	// Leader reports whether this member held the leader lease then.
	Leader bool `json:"leader"`
	// Self marks the member reading the list.
	Self bool `json:"self"`
}

Member is one hub process in the cluster.

func (Member) Endpoint 

func (m Member) Endpoint(name string) string

Endpoint returns a named endpoint, or "".

type Node 

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

Node is this process's membership in the cluster.

func Join 

func Join(opts Options) (*Node, error)

Join registers this process as a member of the control plane at opts.DBPath. The returned Node is not yet heartbeating; call Start.

func (*Node) Adopt 

func (n *Node) Adopt(o Owner, meta any) (bool, error)

Adopt takes over an orphan, conditioned on the claim being exactly the one o was read from. false means another member got there first, or the owner came back.

func (*Node) Assert 

func (n *Node) Assert(kind, key string, meta any) error

Assert records this member as the owner of (kind, key), whoever held it.

func (*Node) BusStats 

func (n *Node) BusStats() BusStats

BusStats returns this member's bus counters.

func (*Node) Call 

func (n *Node) Call(ctx context.Context, to Member, method, path string, in, out any) error

Call makes a JSON request to a peer's internal API: in is sent as the body (nil for none) and a 2xx response is decoded into out (nil to discard). A member judged unresponsive is not asked: Call fails at once with an error wrapping ErrMemberUnresponsive (Task 20419).

func (*Node) Claim 

func (n *Node) Claim(kind, key string, meta any) (Owner, bool, error)

Claim takes (kind, key) for this member unless a live member holds it, in which case it returns that owner and false. A key held by a dead member is taken over. A key this member already holds has its meta refreshed.

func (*Node) Close 

func (n *Node) Close() error

Close leaves the cluster: leadership is released, the member row is marked left so peers adopt what this member owned at once rather than after the TTL, and pending events are flushed. Idempotent.

func (*Node) DialStream added in v0.0.6 

func (n *Node) DialStream(ctx context.Context, to Member, path string) (net.Conn, http.Header, error)

DialStream opens a WebSocket to a peer's internal endpoint, signed as this member, and returns it as a byte stream (Task 20430): how a member that needs a device stream another member holds — a device forwarder's tunnel lives on one member, the agent it is relayed to may be on another — reaches it over the peer channel. path carries its query. The response headers are returned beside the stream, for what the owner reports about it.

The stream is not tied to ctx beyond the dial: it lives until either side closes it, which is what a relay held for a run needs.

func (*Node) Drop 

func (n *Node) Drop(o Owner) (bool, error)

Drop deletes an orphan that nothing can adopt, under the same condition as Adopt.

func (*Node) FlushBus 

func (n *Node) FlushBus()

FlushBus writes queued events now. Tests use it to make publication deterministic; production relies on the flusher.

func (*Node) Forward 

func (n *Node) Forward(w http.ResponseWriter, r *http.Request, to Member, clientIP string) error

Forward hands r to member to and copies its response to w, WebSocket upgrades included. clientIP is the caller as this member resolved it; the owner uses it instead of this member's address. It writes a 502 itself when the member cannot be reached, and returns the error so the caller can log.

func (*Node) ForwardAs 

func (n *Node) ForwardAs(w http.ResponseWriter, r *http.Request, to Member, c Client) error

ForwardAs is Forward with everything the forwarding member knows about the client.

func (*Node) HasPeers 

func (n *Node) HasPeers() bool

HasPeers reports whether any other member is alive. A node without peers skips publishing: nobody would read it.

func (*Node) ID 

func (n *Node) ID() string

ID is this member's instance id.

func (*Node) IsAlive 

func (n *Node) IsAlive(id string) bool

IsAlive reports whether id names a live member. This member is always alive to itself. An id not seen at the last refresh is re-checked against the database, because a member that joined a moment ago is alive whether or not this node has refreshed since.

func (*Node) IsLeader 

func (n *Node) IsLeader() bool

IsLeader reports whether this member currently holds leadership.

func (*Node) LeaderID 

func (n *Node) LeaderID() string

LeaderID returns the instance id last observed holding leadership, or "".

func (*Node) LeadsAdoption 

func (n *Node) LeadsAdoption() bool

LeadsAdoption reports whether this member adopts what members that stopped serving left behind — their container, Pod and host-process runs, the workloads their dispatches started: the leader, which already does the work that must happen once — or, while the leader itself heartbeats without answering, the member serving longest in its stead (Task 20419). An unresponsive leader keeps its lease, which is renewed in the database, so leadership alone would leave its own runs adopted by nobody.

func (*Node) LiveMembers 

func (n *Node) LiveMembers() []Member

LiveMembers returns the members judged alive at the last refresh, this one included.

func (*Node) Lookup 

func (n *Node) Lookup(kind, key string) (o Owner, found bool, err error)

Lookup reports who owns (kind, key). found is false when nobody does.

func (*Node) MarkSeen 

func (n *Node) MarkSeen(kind, key string, expires time.Time) (bool, error)

MarkSeen records a single-use value as seen until expires, cluster-wide, and reports whether it was fresh. See statedb.MarkHubSeen.

func (*Node) Member 

func (n *Node) Member(id string) (Member, bool)

Member returns a member by id, alive or not.

func (*Node) Members 

func (n *Node) Members() []Member

Members returns every member row read at the last refresh, live or not, ordered by start time.

func (*Node) OnBusGap 

func (n *Node) OnBusGap(fn func())

OnBusGap registers fn to run when this member discovers it missed events.

func (*Node) OnLeadership 

func (n *Node) OnLeadership(fn func(leader bool))

OnLeadership registers fn to run on every election (true) and demotion (false). It runs on the campaign goroutine and must not block.

func (*Node) OnMembership 

func (n *Node) OnMembership(fn func(live []Member))

OnMembership registers fn to run after every refresh that changed which members are alive, and after every probe that changed which of them answer (Task 20419). fn receives the members that serve — alive and not judged unresponsive — this one included. It runs on the heartbeat or the probe goroutine and must not block.

func (*Node) Orphans 

func (n *Node) Orphans(kind string) ([]Owner, error)

Orphans lists the rows of kind whose owner is not alive.

func (*Node) Owned 

func (n *Node) Owned(kind string) ([]Owner, error)

Owned lists every row of kind ("" for all kinds), with liveness verdicts.

func (*Node) Owns 

func (n *Node) Owns(kind, key string) bool

Owns reports whether this member currently owns (kind, key), read from the database rather than remembered: ownership can be taken by a member that judged this one dead, and the only way to learn that is to look.

func (*Node) Peers 

func (n *Node) Peers() []Member

Peers returns the live members other than this one.

func (*Node) PollBus 

func (n *Node) PollBus()

PollBus reads and delivers pending events now, for the same reason.

func (*Node) ProbePeers 

func (n *Node) ProbePeers(ctx context.Context)

ProbePeers probes every live peer once, concurrently, and applies the results. The probe loop calls it each interval; a test may call it to move the verdict along without waiting for one.

func (*Node) Publish 

func (n *Node) Publish(topic, key string, payload any)

Publish sends an event to every other member. It is dropped without a write when no other member is alive, because nobody would read it.

func (*Node) PublishTo 

func (n *Node) PublishTo(target, topic, key string, payload any)

PublishTo sends an event to one member. Unlike Publish it is written even when that member is not currently judged alive: a member that joined since the last refresh is alive whether or not this node has noticed.

func (*Node) Release 

func (n *Node) Release(kind, key string) (bool, error)

Release gives up (kind, key) if this member still owns it.

func (*Node) Responsive 

func (n *Node) Responsive(id string) bool

Responsive reports whether this member considers id answering: true for itself, for a member it has not probed, and for one it has heard from since its last failure; false only for a member judged unresponsive.

func (*Node) Self 

func (n *Node) Self() Member

Self returns this member's own record.

func (*Node) Serving 

func (n *Node) Serving(id string) bool

Serving reports whether id is a member that serves this control plane now: heartbeating (IsAlive) and answering (Responsive). It is the test everything that hands work to another member, or takes work over from one, applies.

func (*Node) Start 

func (n *Node) Start(ctx context.Context)

Start begins heartbeating, campaigning for leadership and reading the bus. It returns immediately; Close stops everything.

func (*Node) Subscribe 

func (n *Node) Subscribe(topic string, fn func(Event))

Subscribe registers fn for topic; "*" receives every topic. fn runs on the bus goroutine, in sequence order, and must not block: a slow subscriber delays every event behind it.

func (*Node) Terms 

func (n *Node) Terms() int

Terms reports how many times this member has been elected.

func (*Node) UpdateMeta 

func (n *Node) UpdateMeta(kind, key string, meta any) (bool, error)

UpdateMeta rewrites the meta of a key this member still owns, and reports whether it still did.

func (*Node) VerifyPeer 

func (n *Node) VerifyPeer(r *http.Request) (PeerCall, error)

VerifyPeer checks a request's peer signature. It returns ErrNotPeer for a request that makes no peer claim, and another error for one whose claim is false — which the caller must refuse, not treat as an ordinary client.

func (*Node) WhileLeader 

func (n *Node) WhileLeader(name string, fn func(ctx context.Context))

WhileLeader runs fn, in its own goroutine, for as long as this member leads: started at each election and cancelled at each demotion. Duties registered while the member already leads start at once.

fn must return when its context ends. A duty that ignores cancellation holds up the hand-over for leaderStepDownWait and is then abandoned — with the lease released, so the next leader may start the same duty while this one is still running it. Every duty is written to be safe against that overlap anyway; the wait only makes it rare.

type Options 

type Options struct {
	// DBPath is the control-plane database. Ignored when Store is set.
	DBPath string
	// Store overrides DBPath; the caller then owns its lifetime.
	Store Store

	// Address is the listen address, for display (":8080").
	Address string
	// AdvertiseURL is how peers reach this member's HTTP listener. Required
	// for forwarding: a member with none can serve, but a request another
	// member owns cannot be forwarded to it.
	AdvertiseURL string
	// Version is this build's identifier.
	Version string
	// Endpoints are further named endpoints recorded in the member row's
	// meta: URLs a peer dials, or — for a reader on the same machine, like
	// the Claude home root an offboarding compares (Task 20400) — a path.
	Endpoints map[string]string

	Heartbeat    time.Duration
	MemberTTL    time.Duration
	BusPoll      time.Duration
	BusFlush     time.Duration
	BusRetention time.Duration
	Campaign     time.Duration
	// LeaderTTL and LeaderInterval configure the leader lease. Zero uses the
	// hublease defaults.
	LeaderTTL      time.Duration
	LeaderInterval time.Duration

	// Peer configures the channel members forward requests over.
	Peer PeerOptions

	// ProbeInterval, when positive, has this member ask every live peer over
	// the peer channel whether it is answering, that often (Task 20419; see
	// probe.go). Zero disables probing, which is right only where peers serve
	// no HTTP at all — this package's own tests — since every peer would then
	// be judged unresponsive.
	ProbeInterval time.Duration
	// ProbeTimeout bounds one probe. Zero is DefaultProbeTimeout.
	ProbeTimeout time.Duration
	// UnresponsiveAfter is how long a peer must fail every probe before it is
	// judged unresponsive. Zero is MemberTTL: a member that heartbeats but
	// does not answer is treated as down as soon as a crashed one would be.
	UnresponsiveAfter time.Duration

	// Logf receives operational messages. Nil discards them.
	Logf func(format string, args ...any)

	// Test hooks. Production leaves them zero.
	Now          func() time.Time
	ProcessAlive func(pid int) bool
	Identity     hublease.Identity
	InstanceID   string
}

Options configures Join. DBPath (or Store) is required.

type Owner 

type Owner struct {
	Kind string
	Key  string
	// InstanceID is the owning member's id.
	InstanceID string
	// Member is the owning member as of the last refresh; zero when the row
	// names an instance this node has never seen.
	Member Member
	// Self reports that this member is the owner.
	Self bool
	// Alive reports whether the owner serves: heartbeating, and not judged
	// unresponsive by this member (Serving). An owner that heartbeats but
	// does not answer is not alive here (Task 20419): what it holds is
	// adopted and nothing is forwarded to it, as if it had died.
	Alive bool
	// Row is the stored claim, needed to adopt or drop it.
	Row statedb.HubOwnerRow
}

Owner describes who holds (Kind, Key).

func PeekRunClaims 

func PeekRunClaims(dbPath string, now time.Time) ([]Owner, error)

PeekRunClaims returns every run claim recorded in the control plane at dbPath, each with its owner judged from this process the way members judge each other (RowAlive). For processes that are not members: it reads through a read-only handle, so it neither migrates the hub's database nor is refused by one whose schema is ahead of this binary, and it never writes. A control plane no clustered hub has served has no claims.

func (Owner) ClaimedAt 

func (o Owner) ClaimedAt() time.Time

ClaimedAt is when the current owner took the key.

func (Owner) Meta 

func (o Owner) Meta(v any) error

Meta decodes the owner's recorded metadata into v.

type PeerCall 

type PeerCall struct {
	From     Member
	ClientIP string
}

PeerCall describes a verified peer request.

func PeerCallFrom 

func PeerCallFrom(ctx context.Context) (PeerCall, bool)

PeerCallFrom returns the verified peer call ctx carries, if any. A handler that sees one must answer locally: forwarding it again could loop.

type PeerOptions 

type PeerOptions struct {
	// RootCAs verifies https advertise URLs. Nil uses the system pool.
	RootCAs *x509.CertPool
	// ServerName overrides the name certificates are verified against, for
	// members advertised by IP whose certificate names the public host.
	ServerName string
	// Timeout bounds a Call. Forwarded requests are not bounded as a whole:
	// they may be WebSockets or event streams that live as long as the client
	// wants.
	Timeout time.Duration
	// HeaderTimeout bounds how long a forwarded request waits for the owner's
	// response headers (Task 20419): how long a caller can be held by a member
	// that accepted the request and never answers it. Once the headers arrive
	// the body streams for as long as it lasts. Zero is
	// DefaultPeerHeaderTimeout.
	HeaderTimeout time.Duration
}

PeerOptions configures the forwarding channel.

type RunClaimMeta 

type RunClaimMeta struct {
	// Dispatching marks a claim taken before its workload is on record. A
	// member that dies inside that window leaves a claim the leader clears
	// rather than tries to adopt — once it has stopped whatever workload the
	// dispatch started (RunID, Task 20419).
	Dispatching bool `json:"dispatching,omitempty"`
	// RunID names the run the dispatch is starting, set as soon as its
	// executor is asked (Task 20419). A dispatching claim with one may have a
	// workload behind it, which the dispatch intent of that run names.
	RunID string `json:"run_id,omitempty"`
}

RunClaimMeta is the part of a run claim's metadata every observer judges by. The hub's own run metadata embeds it, so the field is written and read under one name.

func (RunClaimMeta) MayHaveWorkload 

func (m RunClaimMeta) MayHaveWorkload() bool

MayHaveWorkload reports whether a claim may have a workload behind it: one past dispatch, or one whose dispatch got as far as its executor.

type Store 

type Store interface {
	hublease.Store

	JoinHubMember(row statedb.HubMemberRow) error
	HeartbeatHubMember(instanceID string, at time.Time) (bool, error)
	LeaveHubMember(instanceID string, at time.Time) error
	ListHubMembers() ([]statedb.HubMemberRow, error)
	PruneHubMembers(cutoff time.Time) (int64, error)

	AppendHubEvents(events []statedb.HubEventRow) (int64, error)
	HubEventsAfter(after int64, limit int) ([]statedb.HubEventRow, error)
	HubEventBounds() (lowest, highest, issued int64, err error)
	PruneHubEvents(cutoff time.Time) (int64, error)

	GetHubOwner(kind, key string) (statedb.HubOwnerRow, error)
	ListHubOwners(kind string) ([]statedb.HubOwnerRow, error)
	ClaimHubOwner(row, expect statedb.HubOwnerRow) (bool, error)
	PutHubOwner(row statedb.HubOwnerRow) error
	UpdateHubOwnerMeta(kind, key, instanceID, meta string, at time.Time) (bool, error)
	ReleaseHubOwner(kind, key, instanceID string) (bool, error)
	ReleaseHubOwnerIfClaim(expect statedb.HubOwnerRow) (bool, error)

	HubMeta(key string) (string, bool, error)
	SetHubMeta(key, value string) error
	SetHubMetaIfAbsent(key, value string) (string, error)

	MarkHubSeen(kind, key string, expires, now time.Time) (bool, error)
	PruneHubSeen(now time.Time) (int64, error)
}

Store is the persistence a Node needs. *statedb.DB satisfies it.

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.