forge_runtime

package
v0.57.2 Latest Latest
Warning

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

 Go to latest Published: Sep 5, 2026 License: Apache-2.0 Imports: 18 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	// CleanupReasonStop records a caller-requested stop of a live runtime.
	CleanupReasonStop = "stopped"
	// CleanupReasonExpired records a lease-expiry release.
	CleanupReasonExpired = "expired"
)

Cleanup reasons recorded on receipts.

View Source 
const DefaultLeaseDuration = 10 * time.Minute

DefaultLeaseDuration is the reservation lease applied when the admission owner does not configure a duration.

View Source 
const DefaultOwnerLeaseDuration = time.Minute

DefaultOwnerLeaseDuration is the owner claim lease applied when the admission owner does not configure a duration.

Variables

View Source 
var (
	// ErrReservationNotFound is returned when a reservation object key is unknown.
	ErrReservationNotFound = errors.New("reservation not found")
	// ErrWorkerNotObserved is returned when a Worker has no observed capacity record.
	ErrWorkerNotObserved = errors.New("worker capacity not observed")
	// ErrStaleGeneration is returned when a call fences against an older
	// generation, for example a late return from a replaced runtime.
	ErrStaleGeneration = errors.New("stale reservation generation")
	// ErrCapacityExhausted is returned when a worker cannot satisfy a request.
	ErrCapacityExhausted = errors.New("worker capacity exhausted")
	// ErrBackendUnsupported is returned when a worker does not declare the backend.
	ErrBackendUnsupported = errors.New("backend unsupported by worker")
	// ErrReservationTerminal is returned when an idempotent retry hits a
	// released reservation; a retry after release is a new attempt with a new
	// Execution object key.
	ErrReservationTerminal = errors.New("reservation already released")
	// ErrRequestMismatch is returned when an existing reservation conflicts with the request.
	ErrRequestMismatch = errors.New("reservation request mismatch")
	// ErrReservationExpired is returned when an idempotent retry hits a live
	// reservation whose lease already expired but is not swept yet. Run the
	// expiry sweep; the retry then requires a new attempt.
	ErrReservationExpired = errors.New("reservation lease expired")
	// ErrCapacityUnowned is returned when a capacity record carries no live
	// owner claim, including legacy ownerless records.
	ErrCapacityUnowned = errors.New("worker capacity has no live owner claim")
	// ErrCapacityOwned is returned when a different Device holds the live
	// owner claim on a capacity record.
	ErrCapacityOwned = errors.New("worker capacity claimed by another device")
	// ErrCapacityOwnerExpired is returned when the live claim's lease expired
	// without renewal or reclaim.
	ErrCapacityOwnerExpired = errors.New("worker capacity owner claim expired")
	// ErrCapacityDraining is returned when the record's claim is live but in
	// the draining state and cannot accept new work.
	ErrCapacityDraining = errors.New("worker capacity is draining")
)

Errors returned by runtime admission.

Functions

func ApplyDockerWorkdirBind 

func ApplyDockerWorkdirBind(conf *forge_lib_docker.Config, hostPath, containerPath string) error

ApplyDockerWorkdirBind adds the single POSIX bind-mount entry for one supervised Workdir host path to a Docker config.

This adapter does not fence writes: a Docker bind mount lets the container write the host path directly, bypassing any FSHandle writer. The missing piece is a supervised POSIX live-FSHandle mount - a FUSE supervisor backed by db/unixfs/mount MountController exposing the Workdir FSHandle at the host path - owned by the Spacewave filesystem stack. Until that supervisor supplies the host path and its flush, Docker-backed attempts must collect diff evidence only after their own supervisor proves quiescence; this package provides no flush contract for them.

func BuildReservationObjectKey 

func BuildReservationObjectKey(executionObjectKey string) string

BuildReservationObjectKey builds the deterministic object key for the one reservation of an Execution attempt. A retry after release is a new attempt with a new Execution object key.

func BuildWorkerCapacityObjectKey 

func BuildWorkerCapacityObjectKey(workerObjectKey string) string

BuildWorkerCapacityObjectKey builds the deterministic capacity object key for one Forge Worker. The Worker daemon owns this record; Forge validates and debits it, callers never keep a parallel ledger.

func ForgeWorkerObjectKey 

func ForgeWorkerObjectKey(capability *s4wave_device.DeviceCapability) string

ForgeWorkerObjectKey returns the linked Worker object key for a Forge Worker capability, empty when the capability has no linked Worker.

func NewReservationBlock 

func NewReservationBlock() block.Block

NewReservationBlock constructs an empty Reservation block.

func NewWorkerCapacityBlock 

func NewWorkerCapacityBlock() block.Block

NewWorkerCapacityBlock constructs an empty WorkerCapacity block.

func SelectForgeWorkerCapability 

func SelectForgeWorkerCapability(dev *s4wave_device.Device) *s4wave_device.DeviceCapability

SelectForgeWorkerCapability resolves one enrolled Device to its selectable Forge Worker capability using the existing typed selection APIs. Device selection changes the Worker that runs the same manifest; it never creates a Device-specific session path or a second scheduler.

Types

type BackendRuntimeIdentity 

type BackendRuntimeIdentity struct {
	// Backend names the runtime backend that owns the runtime.
	Backend string
	// ID is the backend-scoped runtime identifier, for example a container id.
	ID string
}

BackendRuntimeIdentity identifies one backend runtime instance for one reservation generation. The identity is stable across daemon restarts so reconcile can resume observation without another launch.

func (BackendRuntimeIdentity) IsZero 

func (i BackendRuntimeIdentity) IsZero() bool

IsZero reports whether the identity is unset.

type CapacityOwnerState 

type CapacityOwnerState uint8

CapacityOwnerState describes the lifecycle state of a capacity record's durable owner claim. 

const (
	// CapacityOwnerStateUnspecified is the zero value: no live owner claim.
	// Legacy records written before owner claims decode with this state and
	// stay unavailable to every gated operation.
	CapacityOwnerStateUnspecified CapacityOwnerState = iota
	// CapacityOwnerStateActive means the claim is live and admits new work.
	CapacityOwnerStateActive
	// CapacityOwnerStateDraining means the claim is live but blocks new work
	// until reserved debits clear or declared totals grow to fit them.
	CapacityOwnerStateDraining
)

func (CapacityOwnerState) Valid 

func (s CapacityOwnerState) Valid() bool

Valid reports whether the state is a defined value.

type CleanupReceipt 

type CleanupReceipt struct {
	// ReservationObjectKey is the released reservation object key.
	ReservationObjectKey string
	// ExecutionObjectKey is the owning Execution object key.
	ExecutionObjectKey string
	// RuntimeIdentity is the stopped runtime identity, empty when no runtime launched.
	RuntimeIdentity string
	// Generation is the fenced generation the receipt applies to.
	Generation uint64
	// RuntimeStopped records that the backend runtime was confirmed stopped,
	// or that no runtime ever launched. It stays false while a stop is pending;
	// the receipt never fabricates this fact.
	RuntimeStopped bool
	// CapacityReleased records that reserved capacity was credited back exactly once.
	CapacityReleased bool
	// Reason records why the reservation released: "stopped" or "expired".
	Reason string
}

CleanupReceipt records the terminal cleanup facts for one reservation generation.

func (*CleanupReceipt) Complete 

func (r *CleanupReceipt) Complete() bool

Complete reports whether every cleanup fact is recorded.

func (*CleanupReceipt) MarshalJSON 

func (r *CleanupReceipt) MarshalJSON() ([]byte, error)

MarshalJSON marshals the CleanupReceipt to JSON without reflection.

func (*CleanupReceipt) UnmarshalJSON 

func (r *CleanupReceipt) UnmarshalJSON(data []byte) error

UnmarshalJSON unmarshals the CleanupReceipt from JSON without reflection.

func (*CleanupReceipt) Validate 

func (r *CleanupReceipt) Validate() error

Validate validates the receipt.

type OwnedWorkerCapacity 

type OwnedWorkerCapacity struct {
	// WorkerObjectKey is the Forge Worker object key of the record.
	WorkerObjectKey string
	// Capacity is the owned capacity record.
	Capacity *WorkerCapacity
}

OwnedWorkerCapacity pairs an owned capacity record with the Forge Worker object key it describes, so scans can name workers for reclaim.

type Reservation 

type Reservation struct {
	// WorkerObjectKey is the Forge Worker object key holding the capacity.
	WorkerObjectKey string `json:"workerObjectKey,omitempty"`
	// ExecutionObjectKey is the owning Execution attempt object key.
	ExecutionObjectKey string `json:"executionObjectKey,omitempty"`
	// Request is the reserved capacity request.
	Request ResourceRequest `json:"request"`
	// Generation fences runtime custody. Resume after uncertainty increments
	// the generation; calls fenced against an older generation are stale.
	Generation uint64
	// LeaseExpiresAt is when unobserved custody expires. Expiry releases the
	// debited capacity exactly once.
	LeaseExpiresAt *timestamp.Timestamp
	// State is the reservation lifecycle state.
	State ReservationState
	// Runtime identifies the claimed backend runtime, set on activation.
	Runtime BackendRuntimeIdentity
	// Cleanup records the terminal cleanup receipt, set on release.
	Cleanup *CleanupReceipt
}

Reservation records one generation-fenced and lease-fenced capacity grant. The record is the durable truth: a daemon restart reconciles by reading it.

func LookupReservation 

func LookupReservation(ctx context.Context, ws world.WorldState, objKey string) (*Reservation, error)

LookupReservation loads one persisted Reservation or ErrReservationNotFound. The loaded record is validated before use.

func (*Reservation) LeaseExpired 

func (r *Reservation) LeaseExpired(now time.Time) bool

LeaseExpired reports whether the lease is past due at the given time.

func (*Reservation) MarshalBlock 

func (r *Reservation) MarshalBlock() ([]byte, error)

MarshalBlock marshals the block to binary.

func (*Reservation) MarshalJSON 

func (r *Reservation) MarshalJSON() ([]byte, error)

MarshalJSON marshals the Reservation to JSON without reflection.

func (*Reservation) ObjectKey 

func (r *Reservation) ObjectKey() string

ObjectKey returns the deterministic reservation object key.

func (*Reservation) Outcome 

func (r *Reservation) Outcome(now time.Time) ReservationOutcome

Outcome classifies the reservation for reconcile at the given time.

func (*Reservation) Reset 

func (r *Reservation) Reset()

Reset resets the block.

func (*Reservation) UnmarshalBlock 

func (r *Reservation) UnmarshalBlock(data []byte) error

UnmarshalBlock unmarshals the block from binary.

func (*Reservation) UnmarshalJSON 

func (r *Reservation) UnmarshalJSON(data []byte) error

UnmarshalJSON unmarshals the Reservation from JSON without reflection.

func (*Reservation) Validate 

func (r *Reservation) Validate() error

Validate validates the reservation.

type ReservationOutcome 

type ReservationOutcome uint8

ReservationOutcome classifies a reservation for reconciliation after a daemon restart or a Device reconnect. 

const (
	// OutcomeActive means the reservation holds capacity and custody is fenced.
	OutcomeActive ReservationOutcome = iota + 1
	// OutcomeUncertain means the runtime outcome is unknown: custody went
	// unreachable or a confirmed stop is still pending reconciliation.
	OutcomeUncertain
	// OutcomeTerminal means cleanup is recorded and no capacity is held.
	OutcomeTerminal
)

type ReservationState 

type ReservationState uint8

ReservationState describes whether capacity remains held and whether the runtime outcome is known. 

const (
	// ReservationStateReserved holds debited capacity before a runtime claims it.
	ReservationStateReserved ReservationState = iota + 1
	// ReservationStateActive holds debited capacity for one claimed runtime generation.
	ReservationStateActive
	// ReservationStateUncertain holds debited capacity while the runtime outcome
	// is unknown, for example after a Device disconnect. Capacity stays debited
	// until the same fenced runtime reconnects or the lease expires.
	ReservationStateUncertain
	// ReservationStatePendingStop holds a fenced runtime whose stop has not
	// been confirmed yet. Capacity release follows the expiry rule while the
	// runtime outcome stays unknown until the stop is confirmed.
	ReservationStatePendingStop
	// ReservationStateReleased is terminal: cleanup is recorded and no
	// capacity is held.
	ReservationStateReleased
)

func (ReservationState) Live 

func (s ReservationState) Live() bool

Live reports whether the state may still transition before release.

func (ReservationState) Terminal 

func (s ReservationState) Terminal() bool

Terminal reports whether the state releases capacity permanently.

func (ReservationState) Valid 

func (s ReservationState) Valid() bool

Valid reports whether the state is a defined value.

type ResourceRequest 

type ResourceRequest struct {
	// MilliCPU is the requested CPU in milli-cores.
	MilliCPU uint64
	// MemoryBytes is the requested memory in bytes.
	MemoryBytes uint64
	// Backend names the runtime backend required by the attempt.
	Backend string
}

ResourceRequest declares the host capacity and backend required by one execution attempt.

func (ResourceRequest) Validate 

func (r ResourceRequest) Validate() error

Validate validates the request.

type RuntimeAdmission 

type RuntimeAdmission interface {
	// Reserve atomically debits Worker capacity for one Execution attempt.
	// Reserve is idempotent per Execution object key while the reservation is
	// live; a released reservation requires a new attempt.
	Reserve(ctx context.Context, workerObjectKey, executionObjectKey string, request ResourceRequest) (*Reservation, error)
	// LookupReservation loads one persisted reservation. Reconcile after a
	// restart reads the same object and resumes observation without relaunch.
	LookupReservation(ctx context.Context, reservationObjectKey string) (*Reservation, error)
	// StopAndRelease stops the fenced runtime, credits capacity exactly once,
	// and returns the persisted cleanup facts. The caller must present the
	// live owner claim (ref and current owner epoch) of the Worker's capacity
	// record; a deposed or stale instance is rejected before the stopper runs.
	// A stale reservation generation is rejected without touching the current
	// runtime or capacity. Until the stop is confirmed the reservation sits in
	// the durable pending-stop state.
	StopAndRelease(ctx context.Context, ref WorkerClaimRef, ownerEpoch uint64, reservationObjectKey string, generation uint64) (*CleanupReceipt, error)
}

RuntimeAdmission atomically reserves Worker capacity and reconciles fenced backend runtimes. Forge owns this boundary; callers never keep their own capacity ledger.

type RuntimeStopper 

type RuntimeStopper interface {
	// StopRuntime stops the identified runtime.
	// Returns true when the runtime was stopped or was already gone.
	StopRuntime(ctx context.Context, rt BackendRuntimeIdentity) (bool, error)
}

RuntimeStopper stops one backend runtime by identity. The Forge runtime backend owns the mechanics; admission owns the fence and the ledger. Implementations must be idempotent: stopping an already-gone runtime returns true, nil.

type V86WorkdirMount 

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

V86WorkdirMount registers the writable v86fs mount of one Workdir FSHandle in the v86fs relay server serving the VM. Guest writes traverse v86fs into the FSHandle, so Flush fences them with an engine durability barrier and Release revokes guest access by removing the mount.

func NewV86WorkdirMount 

func NewV86WorkdirMount(
	eng world.Engine,
	server *unixfs_v86fs.Server,
	name, guestPath string,
	handle *unixfs.FSHandle,
) (*V86WorkdirMount, error)

NewV86WorkdirMount constructs the writable v86fs adapter for one Workdir FSHandle. Attach registers it once; a second Attach is rejected because one attempt mounts its Workdir exactly once.

func (*V86WorkdirMount) Attach 

func (m *V86WorkdirMount) Attach() error

Attach registers the writable Workdir mount with the v86fs server exactly once.

func (*V86WorkdirMount) Flush 

func (m *V86WorkdirMount) Flush(ctx context.Context) error

Flush implements WorkdirMount by running the engine durability barrier over every FSHandle write the guest made.

func (*V86WorkdirMount) Release 

func (m *V86WorkdirMount) Release(_ context.Context) error

Release implements WorkdirMount by removing the mount, revoking guest access.

type WorkdirMount 

type WorkdirMount interface {
	// Flush fences pending FSHandle writes into durable storage before diff evidence.
	Flush(ctx context.Context) error
	// Release revokes guest access and tears down the backend mount.
	Release(ctx context.Context) error
}

WorkdirMount is the single writer-fenced live mount of one Workdir FSHandle into one runtime backend. One attempt owns exactly one mount: guest writes enter the Workdir through the Spacewave-owned FSHandle writer, so the FSHandle is the only write fence. Flush fences every write that traversed the FSHandle into durable storage; call it once before diff evidence.

type WorkerCapacity 

type WorkerCapacity struct {
	// WorkerObjectKey is the Forge Worker object key this record describes.
	// The record's own object key is a one-way hash of this value, so scans
	// and reclaim need it stored to name the Worker.
	WorkerObjectKey string
	// MilliCPUTotal is the observed total CPU in milli-cores.
	MilliCPUTotal uint64
	// MilliCPUReserved is the CPU currently debited by live reservations.
	MilliCPUReserved uint64
	// MemoryBytesTotal is the observed total memory in bytes.
	MemoryBytesTotal uint64
	// MemoryBytesReserved is the memory currently debited by live reservations.
	MemoryBytesReserved uint64
	// Backends lists the runtime backends the Worker supports.
	Backends []string
	// ObservedAt is when the Worker reported the totals.
	ObservedAt *timestamp.Timestamp
	// Generation increments on every mutation of this record so observers can
	// fence stale capacity views.
	Generation uint64
	// OwnerDeviceObjectKey is the enrolled Device object key of the owning
	// daemon instance. Empty together with every other owner field marks a
	// legacy ownerless record.
	OwnerDeviceObjectKey string
	// ClaimID identifies one owning daemon instance claim.
	ClaimID string
	// OwnerEpoch increments on every claim transition; calls fencing against
	// an older epoch are stale.
	OwnerEpoch uint64
	// OwnerLeaseExpiresAt is when an unrenewed owner claim expires.
	OwnerLeaseExpiresAt *timestamp.Timestamp
	// OwnerState is the claim lifecycle state.
	OwnerState CapacityOwnerState
}

WorkerCapacity is the observed and reserved capacity for one Forge Worker.

func LookupWorkerCapacity 

func LookupWorkerCapacity(ctx context.Context, ws world.WorldState, workerObjectKey string) (*WorkerCapacity, error)

LookupWorkerCapacity loads one Worker capacity record or ErrWorkerNotObserved.

func (*WorkerCapacity) MarshalBlock 

func (w *WorkerCapacity) MarshalBlock() ([]byte, error)

MarshalBlock marshals the block to binary.

func (*WorkerCapacity) MarshalJSON 

func (w *WorkerCapacity) MarshalJSON() ([]byte, error)

MarshalJSON marshals the WorkerCapacity to JSON without reflection.

func (*WorkerCapacity) OwnerClaimActive 

func (w *WorkerCapacity) OwnerClaimActive(now time.Time) error

OwnerClaimActive reports whether the record carries a live owner claim at the given time: owned, lease unexpired, and state ACTIVE. It returns typed sentinel errors so callers can distinguish unavailable, expired, and draining records without string matching.

func (*WorkerCapacity) Reset 

func (w *WorkerCapacity) Reset()

Reset resets the block.

func (*WorkerCapacity) SupportsBackend 

func (w *WorkerCapacity) SupportsBackend(backend string) bool

SupportsBackend reports whether the Worker declares the backend.

func (*WorkerCapacity) UnmarshalBlock 

func (w *WorkerCapacity) UnmarshalBlock(data []byte) error

UnmarshalBlock unmarshals the block from binary.

func (*WorkerCapacity) UnmarshalJSON 

func (w *WorkerCapacity) UnmarshalJSON(data []byte) error

UnmarshalJSON unmarshals the WorkerCapacity from JSON without reflection.

func (*WorkerCapacity) Validate 

func (w *WorkerCapacity) Validate() error

Validate validates the capacity record. A record carries either no owner fields at all (legacy ownerless shape) or all six; partial owner shapes are invalid so a half-written claim can never decode as available.

type WorkerClaimRef 

type WorkerClaimRef struct {
	// DeviceObjectKey is the enrolled Device object key of the instance.
	DeviceObjectKey string
	// ClaimID is the per-instance claim identifier. A new claim id on the same
	// Device replaces a previous instance after reclaim.
	ClaimID string
}

WorkerClaimRef identifies one daemon instance's durable owner claim on a capacity record. The reference is presented by the caller and checked against the record's stored claim inside each write transaction.

type WorldRuntimeAdmission 

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

WorldRuntimeAdmission implements RuntimeAdmission over durable world objects.

Every reservation and capacity mutation applies inside one world write transaction, so create-plus-debit and release-plus-credit are atomic even across a crash. All mutations for one Worker serialize on a per-Worker lock; the durable boundary assumes exactly one writer instance per Forge Worker capacity record. A second daemon writing the same Worker's capacity record is outside this contract and must not be attempted: run one admission owner per Worker.

func NewWorldRuntimeAdmission 

func NewWorldRuntimeAdmission(eng world.Engine, stopper RuntimeStopper, lease, ownerLease time.Duration) *WorldRuntimeAdmission

NewWorldRuntimeAdmission constructs a world-backed RuntimeAdmission. A zero lease uses DefaultLeaseDuration; a zero ownerLease uses DefaultOwnerLeaseDuration.

func (*WorldRuntimeAdmission) Activate 

func (a *WorldRuntimeAdmission) Activate(
	ctx context.Context,
	reservationObjectKey string,
	rt BackendRuntimeIdentity,
) (*Reservation, error)

Activate claims the reserved capacity for one backend runtime before launch.

func (*WorldRuntimeAdmission) BeginDrainCapacity 

func (a *WorldRuntimeAdmission) BeginDrainCapacity(
	ctx context.Context,
	workerObjectKey string,
	ref WorkerClaimRef,
	epoch uint64,
) (*WorkerCapacity, error)

BeginDrainCapacity moves the record to DRAINING with empty backends under a live claim at the given epoch. Reserved debits stay held; sweeps continue. Idempotent.

func (*WorldRuntimeAdmission) ClaimWorkerCapacity 

func (a *WorldRuntimeAdmission) ClaimWorkerCapacity(
	ctx context.Context,
	workerObjectKey string,
	ref WorkerClaimRef,
) (*WorkerCapacity, error)

ClaimWorkerCapacity claims or reclaims the Worker's capacity record for the calling instance. An absent record is created at epoch 1 with zero totals; a legacy ownerless record or an expired lease is reclaimed with an epoch bump that preserves OwnerState so resumed drains stay draining. A live foreign Device claim fails with ErrCapacityOwned. The same Device and claim id renew idempotently without bumping the epoch.

func (*WorldRuntimeAdmission) CompleteDrainCapacity 

func (a *WorldRuntimeAdmission) CompleteDrainCapacity(
	ctx context.Context,
	workerObjectKey string,
	ref WorkerClaimRef,
	epoch uint64,
) error

CompleteDrainCapacity deletes a fully drained capacity record exactly once. It requires the DRAINING state and no non-terminal reservation referencing the Worker; the scan runs inside the same transaction as the deletion.

func (*WorldRuntimeAdmission) ExpireLeases 

func (a *WorldRuntimeAdmission) ExpireLeases(ctx context.Context, ref WorkerClaimRef, now time.Time) ([]*CleanupReceipt, error)

ExpireLeases fences every live reservation whose lease expired at or before now. Each expiry is one transaction that bumps the generation, enters the durable pending-stop state while retaining the debit, and persists the explicitly partial receipt: RuntimeStopped=false because the runtime outcome is unknown, CapacityReleased=false because the debit is held until the stop confirms. After committing, the sweep attempts the stop outside the Worker lock; confirmation finalizes the truthful terminal receipt and credits the debit exactly once. Failure leaves the work to ReconcilePendingStops. Post-expiry calls fenced against the old generation are stale and cannot receive the terminal receipt.

func (*WorldRuntimeAdmission) LookupReservation 

func (a *WorldRuntimeAdmission) LookupReservation(ctx context.Context, reservationObjectKey string) (*Reservation, error)

LookupReservation implements RuntimeAdmission.

func (*WorldRuntimeAdmission) LookupWorkerCapacityAdmission 

func (a *WorldRuntimeAdmission) LookupWorkerCapacityAdmission(ctx context.Context, workerObjectKey string) (*WorkerCapacity, error)

LookupWorkerCapacityAdmission loads one capacity record through the admission instance's read transaction.

func (*WorldRuntimeAdmission) MarkUncertain 

func (a *WorldRuntimeAdmission) MarkUncertain(ctx context.Context, reservationObjectKey string) (*Reservation, error)

MarkUncertain records that runtime custody is unreachable while capacity stays debited, whether before launch or after activation. Idempotent while uncertain; rejected once released or pending stop.

func (*WorldRuntimeAdmission) ObserveWorker 

func (a *WorldRuntimeAdmission) ObserveWorker(
	ctx context.Context,
	workerObjectKey string,
	ref WorkerClaimRef,
	epoch uint64,
	milliCPUTotal, memoryBytesTotal uint64,
	backends []string,
) (*WorkerCapacity, error)

ObserveWorker upserts the Worker's observed capacity totals and backends under a live owner claim at the given epoch. Reserved debits are preserved; every observation bumps the record generation and stamps ObservedAt. Declared totals below current debits move the record to DRAINING until credits land; fitting totals return it to ACTIVE. Empty backends are a validation error: only BeginDrainCapacity empties the backend list.

func (*WorldRuntimeAdmission) ReconcilePendingStops 

func (a *WorldRuntimeAdmission) ReconcilePendingStops(ctx context.Context, ref WorkerClaimRef) ([]*CleanupReceipt, error)

ReconcilePendingStops confirms stops for every pending-stop reservation by running the idempotent stopper and finalizing the receipt. It completes the work of a crashed StopAndRelease or an unreachable expired runtime.

func (*WorldRuntimeAdmission) RenewLease 

func (a *WorldRuntimeAdmission) RenewLease(ctx context.Context, ref WorkerClaimRef, reservationObjectKey string) (*Reservation, error)

RenewLease extends the lease of a live reservation from the owning instance. The claim reference is verified against the worker record's durable live claim inside the transition transaction: a deposed instance renewing leases in a loop must not starve the new owner's expiry sweep.

func (*WorldRuntimeAdmission) RenewWorkerClaim 

func (a *WorldRuntimeAdmission) RenewWorkerClaim(
	ctx context.Context,
	workerObjectKey string,
	ref WorkerClaimRef,
) (*WorkerCapacity, error)

RenewWorkerClaim extends the owner claim lease of one Worker. Renewal on an expired lease falls back to reclaim: the same ref reclaims with an epoch bump and preserved state instead of stalling its own sweeps.

func (*WorldRuntimeAdmission) Reserve 

func (a *WorldRuntimeAdmission) Reserve(
	ctx context.Context,
	workerObjectKey, executionObjectKey string,
	request ResourceRequest,
) (*Reservation, error)

Reserve implements RuntimeAdmission. Creation and the capacity debit apply in one transaction; the idempotent return proves the debit by re-reading the capacity record before returning.

func (*WorldRuntimeAdmission) ResumeFromUncertain 

func (a *WorldRuntimeAdmission) ResumeFromUncertain(
	ctx context.Context,
	reservationObjectKey string,
	rt BackendRuntimeIdentity,
) (*Reservation, error)

ResumeFromUncertain re-fences custody when the same fenced runtime reconnects. The generation increments so a late return from the previous runtime instance cannot stop or release the re-fenced reservation.

func (*WorldRuntimeAdmission) ScanOwnedCapacity 

func (a *WorldRuntimeAdmission) ScanOwnedCapacity(
	ctx context.Context,
	deviceObjectKey string,
) ([]OwnedWorkerCapacity, error)

ScanOwnedCapacity returns the capacity records owned by one Device, paired with their Worker object keys.

func (*WorldRuntimeAdmission) SetTimeNow 

func (a *WorldRuntimeAdmission) SetTimeNow(now func() time.Time)

SetTimeNow overrides the clock; tests use this to drive expiry.

func (*WorldRuntimeAdmission) StopAndRelease 

func (a *WorldRuntimeAdmission) StopAndRelease(
	ctx context.Context,
	ref WorkerClaimRef,
	ownerEpoch uint64,
	reservationObjectKey string,
	generation uint64,
) (*CleanupReceipt, error)

StopAndRelease implements RuntimeAdmission.

The transition into the durable pending-stop state applies under the Worker lock in its own transaction. The runtime stop runs without the lock held. The finalizing transaction persists the confirmed receipt and credits the capacity atomically. A stale generation is rejected without touching the current runtime or capacity: that call belongs to a replaced runtime instance returning late. A crash between transition and finalize leaves the reservation durably in the pending-stop state; ReconcilePendingStops finishes it with the same idempotent stopper.

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.