Documentation
¶
Overview ¶
Package dispatcher holds the dispatcher's application state: the registry of the executors that currently hold a control session, admission of submitted runs against reserved capacity, and the results those runs report.
An executor is in the registry as an entry keyed by its ID, holding the metadata of its session and the transport owner that session belongs to. The owner is the authority: an entry is exposed, and work is admitted for it, only while its owner is active, registered and inside its lease, never because an executor reported a heartbeat. The transport publishes an owner and calls registration under the setup operation it admitted for it; publication into the registry is committed under the owner's own guard, so a retirement racing it wins. Package transport/rpc defines owners, mutations and their lifetime.
Callbacks from the control transport arrive with the mutation they were admitted under, and that mutation's owner is compared against the executor's current owner before any effect of it is applied. Work this package starts itself, uploading or aborting a run on an executor, admits its own mutation on that executor's owner, and a continuation that has to outlive the request which started it is forked from the live mutation rather than detached from it.
Close closes admission first and then cancels and joins what this package owns: the registrations in flight, the lease expiry goroutine and the control transport. Cancelling is not completing — a canceled call is still waited for.
Index ¶
- Constants
- Variables
- func AdmissionPaused() error
- type Dispatcher
- func (d *Dispatcher) AbortDebuglet(ctx context.Context, executorID string, debugletID uuid.UUID, reason string) error
- func (d *Dispatcher) Close()
- func (d *Dispatcher) ControlIncarnation() string
- func (d *Dispatcher) ControlLeaseDuration() time.Duration
- func (d *Dispatcher) ExecutorEligibility() (eligible, registered int)
- func (d *Dispatcher) GetExecutor(id string) (*RegisteredExecutor, bool)
- func (d *Dispatcher) GetExecutorByIPFull(ip string) (RegisteredExecutor, bool)
- func (d *Dispatcher) GetKeyStore() *tag.KeyStore
- func (d *Dispatcher) GetVersion() string
- func (d *Dispatcher) ListExecutors() []RegisteredExecutor
- func (d *Dispatcher) OnDebugletAllocate(ctx context.Context, mutation *rpc.Mutation, req *pb.DebugletAllocateRequest) (*pb.DebugletAllocateResponse, error)
- func (d *Dispatcher) OnDebugletExit(ctx context.Context, mutation *rpc.Mutation, req *pb.DebugletExitRequest) (*pb.DebugletExitResponse, error)
- func (d *Dispatcher) OnDebugletState(ctx context.Context, mutation *rpc.Mutation, req *pb.DebugletStateRequest) (*pb.DebugletStateResponse, error)
- func (d *Dispatcher) OnDebugletStream(owner *rpc.SessionOwner, ...) error
- func (d *Dispatcher) OnExecutorConnected(ctx context.Context, owner *rpc.SessionOwner, hello *pb.HelloResponse, ...) error
- func (d *Dispatcher) OnExecutorDisconnected(owner *rpc.SessionOwner)
- func (d *Dispatcher) OnHeartbeat(ctx context.Context, mutation *rpc.Mutation, req *pb.HeartbeatRequest) (*pb.HeartbeatResponse, error)
- func (d *Dispatcher) OnResources(ctx context.Context, mutation *rpc.Mutation, req *pb.ResourcesRequest) (*pb.ResourcesResponse, error)
- func (d *Dispatcher) RegisterExecutor(ctx context.Context, owner *rpc.SessionOwner, hello *pb.HelloResponse, ...) error
- func (d *Dispatcher) RestoreScheduler(ctx context.Context) error
- func (d *Dispatcher) SetDestinationLimit(destination string, limit resource.Bitrate) error
- func (d *Dispatcher) SubmitDebuglets(ctx context.Context, specs []models.DebugletSpec, userID *uuid.UUID) (uuid.UUIDs, error)
- type MaintenanceSwitch
- type RegisteredExecutor
Constants ¶
const MaintenanceFileEnv = "DEBUGLET_MAINTENANCE_FILE"
MaintenanceFileEnv names the environment variable that points the dispatcher at its maintenance switch. A dispatcher started without it has no switch and performs no file access per submission at all.
Variables ¶
var ( ErrSessionRetired = errors.New("executor session retired") ErrDispatcherClosed = errors.New("dispatcher closed") )
var ErrAbortRefused = errors.New("executor refused the cancellation")
ErrAbortRefused marks a cancellation its executor answered with a refusal, as distinct from a cancellation that did not reach it.
var ErrCancellationNotRecorded = errors.New("cancellation acknowledged but its result was not recorded")
ErrCancellationNotRecorded marks a cancellation its executor acknowledged but whose terminal result the dispatcher failed to record. It is no refusal: the run may already be stopping, while its stored state does not show the cancellation.
var ErrInvalidPolicy = errors.New("invalid policy")
ErrInvalidPolicy marks a submission the dispatcher refuses because of the numbers it carries, rather than because of anything the dispatcher is or does. It is a rejection of the request, so a caller can answer it as one: the HTTP boundary reports it as an invalid policy and not as a failure of the server.
var ErrMaintenanceMode = errors.New("dispatcher is in maintenance: submission admission is stopped")
ErrMaintenanceMode reports a dispatcher whose operator has stopped submission admission. It is a refusal to accept new work, not a failure: nothing was reserved, persisted or charged.
var ErrPaymentInUse = errors.New("payment order already has admitted runs")
ErrPaymentInUse marks a submission refused where the transaction's orders already carry runs, or where the dispatcher could not tell whether they do. Those runs may be executing or already credited to their executor, so the caller must not refund the transaction on such a refusal.
Functions ¶
func AdmissionPaused ¶
func AdmissionPaused() error
AdmissionPaused reports whether this dispatcher may admit new submissions. A transport checks it before it looks up or charges anything, so a refusal costs a submitter nothing; the submission path checks it again, because that is where the decision has to hold.
Types ¶
type Dispatcher ¶
type Dispatcher struct {
Bidi *rpc.BidiServer
Payment *payments.PaymentHandler
// contains filtered or unexported fields
}
func (*Dispatcher) AbortDebuglet ¶
func (d *Dispatcher) AbortDebuglet(ctx context.Context, executorID string, debugletID uuid.UUID, reason string) error
AbortDebuglet asks the run's executor to cancel it and records the cancellation as the run's terminal result. A nil error means that result is recorded, or that the run was already terminal and keeps its own. An error wrapping ErrCancellationNotRecorded means the executor acknowledged the cancellation but its result is not recorded. Other errors do not confirm that the cancellation was recorded.
A run whose control session has ended, because the dispatcher restarted or the executor registered again in a new session, has no executor to ask: its cancellation is recorded locally, see cancelUnbound.
func (*Dispatcher) Close ¶
func (d *Dispatcher) Close()
Close closes admission before it cancels and joins owned work. Resource Close and cancellation hooks never run under either shared map lock.
func (*Dispatcher) ControlIncarnation ¶
func (d *Dispatcher) ControlIncarnation() string
ControlIncarnation is the nonsecret identity of this dispatcher lifetime. A transport installed in place of the one New built carries it unchanged.
func (*Dispatcher) ControlLeaseDuration ¶
func (d *Dispatcher) ControlLeaseDuration() time.Duration
ControlLeaseDuration is the lease policy of this dispatcher lifetime, retained with the incarnation by any such replacement.
func (*Dispatcher) ExecutorEligibility ¶
func (d *Dispatcher) ExecutorEligibility() (eligible, registered int)
ExecutorEligibility counts the registry and the subset of it that may be given work now, which is the entries whose owner is available. It copies no record and reads no database.
func (*Dispatcher) GetExecutor ¶
func (d *Dispatcher) GetExecutor(id string) (*RegisteredExecutor, bool)
func (*Dispatcher) GetExecutorByIPFull ¶
func (d *Dispatcher) GetExecutorByIPFull(ip string) (RegisteredExecutor, bool)
GetExecutorByIPFull selects the latest local observation, with a stable ID tie-break, and returns metadata independent of subsequent registry changes.
func (*Dispatcher) GetKeyStore ¶
func (d *Dispatcher) GetKeyStore() *tag.KeyStore
func (*Dispatcher) GetVersion ¶
func (d *Dispatcher) GetVersion() string
func (*Dispatcher) ListExecutors ¶
func (d *Dispatcher) ListExecutors() []RegisteredExecutor
func (*Dispatcher) OnDebugletAllocate ¶
func (d *Dispatcher) OnDebugletAllocate(ctx context.Context, mutation *rpc.Mutation, req *pb.DebugletAllocateRequest) (*pb.DebugletAllocateResponse, error)
func (*Dispatcher) OnDebugletExit ¶
func (d *Dispatcher) OnDebugletExit(ctx context.Context, mutation *rpc.Mutation, req *pb.DebugletExitRequest) (*pb.DebugletExitResponse, error)
OnDebugletExit handles the exit of a debuglet, cleaning up its state and notifying any connected log streams. It is safe to call multiple times: CompleteDebuglet is the sole terminal writer and its guard admits exactly one winner per debuglet. Only that winner runs the payment decision, destination removal, scheduler release and fairshare update; later callbacks for a terminal row acknowledge with no effects.
func (*Dispatcher) OnDebugletState ¶
func (d *Dispatcher) OnDebugletState(ctx context.Context, mutation *rpc.Mutation, req *pb.DebugletStateRequest) (*pb.DebugletStateResponse, error)
func (*Dispatcher) OnDebugletStream ¶
func (d *Dispatcher) OnDebugletStream(owner *rpc.SessionOwner, stream grpc.BidiStreamingServer[pb.DebugletStreamRequest, pb.DebugletStreamResponse]) error
func (*Dispatcher) OnExecutorConnected ¶
func (d *Dispatcher) OnExecutorConnected(ctx context.Context, owner *rpc.SessionOwner, hello *pb.HelloResponse, sourceIP string) error
func (*Dispatcher) OnExecutorDisconnected ¶
func (d *Dispatcher) OnExecutorDisconnected(owner *rpc.SessionOwner)
func (*Dispatcher) OnHeartbeat ¶
func (d *Dispatcher) OnHeartbeat(ctx context.Context, mutation *rpc.Mutation, req *pb.HeartbeatRequest) (*pb.HeartbeatResponse, error)
func (*Dispatcher) OnResources ¶
func (d *Dispatcher) OnResources(ctx context.Context, mutation *rpc.Mutation, req *pb.ResourcesRequest) (*pb.ResourcesResponse, error)
func (*Dispatcher) RegisterExecutor ¶
func (d *Dispatcher) RegisterExecutor(ctx context.Context, owner *rpc.SessionOwner, hello *pb.HelloResponse, sourceIP string) error
RegisterExecutor stages metadata for an already-published transport owner. ctx belongs to the setup mutation the transport admitted on that owner, so retirement cancels it and a replacement session waits for this call to return. Registration admits no mutation of its own.
func (*Dispatcher) RestoreScheduler ¶
func (d *Dispatcher) RestoreScheduler(ctx context.Context) error
RestoreScheduler reserves again, when the dispatcher starts, the floors of the stored runs whose window has not ended, or ended less than a minute ago, so that admission counts them as the previous dispatcher did. A run whose stored state is exited released its floor when it finished and reserves nothing; every other run, pending or of uncertain outcome, keeps its reservation until its window ends.
A restored run bound to a previous dispatcher lifetime is logged as a warning with its ID: its control session ended with that lifetime, so it will not execute, and its reservation lasts until the run is cancelled or its window ends. A run without a complete stored binding cannot be cancelled; its reservation lasts until its window ends.
Reservations are restored once per dispatcher lifetime: after a restore has succeeded, a further call reserves nothing and returns an error, so no run is counted twice. A call that fails has reserved nothing and may be repeated.
func (*Dispatcher) SetDestinationLimit ¶
func (d *Dispatcher) SetDestinationLimit(destination string, limit resource.Bitrate) error
SetDestinationLimit records the limit of destination and sends the share it recomputes to every executor holding an allocation there, waiting up to five seconds, on a context of its own, for those deliveries. A limit below the floors already charged there is refused with resource.ErrCapacityFull before anything is recorded or sent. Otherwise the limit governs admission at once and stays recorded whether or not every executor acknowledged; the error joins the failed deliveries.
func (*Dispatcher) SubmitDebuglets ¶
func (d *Dispatcher) SubmitDebuglets(ctx context.Context, specs []models.DebugletSpec, userID *uuid.UUID) (uuid.UUIDs, error)
type MaintenanceSwitch ¶
type MaintenanceSwitch struct {
SchemaVersion int `json:"schema_version"`
Paused bool `json:"paused"`
Reason string `json:"reason,omitempty"`
Since string `json:"since,omitempty"`
}
MaintenanceSwitch is the record the switch file carries.
type RegisteredExecutor ¶
type RegisteredExecutor struct {
ID string
Version string
Ready bool
LastSeen time.Time
TeslaDelay time.Duration
TeslaAnchorTimestamp time.Time
TeslaAnchorKey []byte // k_0, the public chain anchor
ICMPEnabled bool
PricePerBwS int64
Currency string
SuiWallet string
// contains filtered or unexported fields
}
RegisteredExecutor represents a registered executor and its metadata.
func (*RegisteredExecutor) AppendDebugletID ¶
func (e *RegisteredExecutor) AppendDebugletID(id uuid.UUID)
AppendDebugletID adds id to the executor's history, trimming old entries so the total length stays within 2× the maximum to bound memory usage.
func (*RegisteredExecutor) PublicHost ¶
func (e *RegisteredExecutor) PublicHost() string
PublicHost returns the executor's public host (IP or domain) at which debuglet listeners can be contacted, or "" if none is configured.
func (*RegisteredExecutor) RecentDebugletIDs ¶
func (e *RegisteredExecutor) RecentDebugletIDs(n int) []uuid.UUID
RecentDebugletIDs returns up to n recent debuglet IDs for this executor, newest first. If n ≤ 0 the default (lastDebugletHistory) is used.
Source Files
¶
Directories
¶
| Path | Synopsis |
|---|---|
|
Package enrollment binds an executor ID to the node credential its machine holds: the SHA-256 fingerprint of the client certificate the dispatcher's configured authority verified on a control connection.
|
Package enrollment binds an executor ID to the node credential its machine holds: the SHA-256 fingerprint of the client certificate the dispatcher's configured authority verified on a control connection. |
|
Package testutil holds test support shared by dispatcher packages.
|
Package testutil holds test support shared by dispatcher packages. |
|
transport
|
|
|
rpc
Package rpc serves the dispatcher's side of the executor control plane and owns one session per connected executor.
|
Package rpc serves the dispatcher's side of the executor control plane and owns one session per connected executor. |