Help us shape the future of Go! Take the 2026 Go Developer Survey

plugin_host_scheduler

package
v0.57.13 Latest Latest
Warning

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

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

README

Plugin Host Scheduler

This controller manages selecting which manifest to execute for each plugin and executing the plugins. It also manages attempting to fully download copies of new plugin versions before replacing the old plugin version instances to avoid interruptions in the User Experience.

Overview of Routines

Here are the routines managed by the scheduler:

Global, for all plugins:

  • Task 0: Watch the list of available plugin hosts with CollectValues (Execute())

Managed by the pluginInstance for each plugin-id we want to execute:

  • Task 1: FetchManifests: call FetchManifest with platform id set to any

    • Unless the DisableFetchManifests option is set.
    • Write the ManifestRefs into the world if newer than the latest (or not exist) in one transaction
    • We will track all available platform manifests for the plugin id in the world,

  • Task 2: SelectManifest: select the best manifest(s) from the set available in the world.

    • In a world transaction: query the list of available manifests for the plugin id
      • search for Manifests or ManifestRefs
    • Drop any manifests that are for platform IDs we don't have hosts for
    • Sort by rev, followed by preferred platform ID to less preferred platform ID
    • Iterate over the list and identify two manifests from the set of available:
      • Download manifest: the first manifest in the list.
      • Live manifest: the first manifest in the list which has been fully downloaded.
      • Fallback to the download manifest for the Live manifest if none are downloaded.
    • Set the Download manifest to the DownloadManifest state routine
    • Set the Live manifest to the ExecutePlugin state routine

  • Task 3: DownloadManifest: download the "Download manifest" to storage fully

    • Traverse all the blocks in the block graph
    • Any block not present in local storage => fetch and copy to local storage
    • Mark the manifest as fully downloaded once complete.

  • Task 5: ExecutePlugin: execute the plugin with a ManifestSnapshot

    • Mount the plugin fs from the world block graph
    • Execute the plugin

Documentation

Index

Constants

View Source 
const ConfigID = ControllerID

ConfigID is the config identifier.

View Source 
const ControllerID = "bldr/plugin/host/scheduler"

ControllerID is the controller identifier.

View Source 
const DefaultStartupWaitBudget = time.Minute

DefaultStartupWaitBudget is the startup wait budget used when the config leaves StartupWaitBudgetDur unset or zero.

Variables

View Source 
var Version = controller.MustParseVersion("0.0.1")

Version is the version of the controller implementation.

Functions

func IsPluginRunningOnBus added in v0.51.5 

func IsPluginRunningOnBus(ctx context.Context, b bus.Bus, pluginID string) bool

IsPluginRunningOnBus reports whether the scheduler on b has a running plugin.

Types

type Config 

type Config struct {

	// EngineId is the world engine id to attach to.
	EngineId string `protobuf:"bytes,1,opt,name=engine_id,json=engineId,proto3" json:"engineId,omitempty"`
	// ObjectKey is the root object to attach to.
	// If not exists, waits for it to exist.
	//
	// Searches for <manifest> links from this object.
	ObjectKey string `protobuf:"bytes,2,opt,name=object_key,json=objectKey,proto3" json:"objectKey,omitempty"`
	// PeerId is the peer ID to use for world transactions.
	PeerId string `protobuf:"bytes,3,opt,name=peer_id,json=peerId,proto3" json:"peerId,omitempty"`
	// VolumeId is the identifier of the volume on the plugin host bus.
	// This volume is available for the plugin to use via the volume proxy.
	VolumeId string `protobuf:"bytes,4,opt,name=volume_id,json=volumeId,proto3" json:"volumeId,omitempty"`
	// WatchFetchManifest will watch the FetchManifest directive while a plugin is running.
	WatchFetchManifest bool `protobuf:"varint,5,opt,name=watch_fetch_manifest,json=watchFetchManifest,proto3" json:"watchFetchManifest,omitempty"`
	// DisableStoreManifest disables storing manifests fetched with FetchManifest.
	// This is used if we are watching the same world as the FetchManifest resolver.
	DisableStoreManifest bool `protobuf:"varint,6,opt,name=disable_store_manifest,json=disableStoreManifest,proto3" json:"disableStoreManifest,omitempty"`
	// FetchConcurrency limits the number of blocks fetched concurrently per-manifest.
	// If zero, uses a finite default of 2, shared by the serialized active
	// manifest copy allowance.
	//
	// Note: the concurrency is limited by the number of blocks that we have seen
	// so far. Fetches blocks, then the references those blocks reference. We only
	// know about the blocks on the frontier of the blocks fetched so far.
	FetchConcurrency uint32 `protobuf:"varint,7,opt,name=fetch_concurrency,json=fetchConcurrency,proto3" json:"fetchConcurrency,omitempty"`
	// FetchBackoff is the backoff config for fetching plugin manifests.
	// If unset, defaults to reasonable defaults.
	FetchBackoff *backoff.Backoff `protobuf:"bytes,8,opt,name=fetch_backoff,json=fetchBackoff,proto3" json:"fetchBackoff,omitempty"`
	// ExecBackoff is the backoff config for executing plugin manifests.
	// If unset, defaults to reasonable defaults.
	ExecBackoff *backoff.Backoff `protobuf:"bytes,9,opt,name=exec_backoff,json=execBackoff,proto3" json:"execBackoff,omitempty"`
	// DisableCopyManifest disables copying manifests to the plugin host world bucket.
	// This is used if the manifest bucket is always accessible and locally cached.
	// If unset, we will copy manifests to the same bucket as the plugin host world.
	DisableCopyManifest bool `protobuf:"varint,10,opt,name=disable_copy_manifest,json=disableCopyManifest,proto3" json:"disableCopyManifest,omitempty"`
	// Verbose enables verbose logging for world ops (slower).
	Verbose bool `protobuf:"varint,11,opt,name=verbose,proto3" json:"verbose,omitempty"`
	// PlatformSelectionPolicies restrict selected platform IDs by plugin ID. If
	// empty, every discovered plugin host platform is selectable for every
	// plugin.
	PlatformSelectionPolicies []*PlatformSelectionPolicy `` /* 137-byte string literal not displayed */
	// NoCopyBucketIds lists source buckets that remain authoritative without a
	// full-DAG copy.
	NoCopyBucketIds []string `protobuf:"bytes,13,rep,name=no_copy_bucket_ids,json=noCopyBucketIds,proto3" json:"noCopyBucketIds,omitempty"`
	// InstanceKey identifies the isolated plugin instance set resolved by this scheduler.
	// Empty retains the unscoped Dist behavior.
	InstanceKey string `protobuf:"bytes,14,opt,name=instance_key,json=instanceKey,proto3" json:"instanceKey,omitempty"`
	// StartupWaitBudgetDur is the maximum wall-clock duration a plugin instance
	// may take to complete initial capability registration before its startup
	// wait budget is reported as exhausted. Execution retries remain unbounded.
	// If unset or zero, defaults to 60s. Negative values are rejected.
	StartupWaitBudgetDur string `protobuf:"bytes,15,opt,name=startup_wait_budget_dur,json=startupWaitBudgetDur,proto3" json:"startupWaitBudgetDur,omitempty"`
	// MaterializerPluginId is the plugin ID to route manifest materialization
	// through. If empty, the scheduler materializes manifests directly.
	MaterializerPluginId string `protobuf:"bytes,16,opt,name=materializer_plugin_id,json=materializerPluginId,proto3" json:"materializerPluginId,omitempty"`
	// UpdateGuardPluginIds require a successful UpdateGuard.Prepare call on the
	// running generation before a different executable replaces it.
	UpdateGuardPluginIds []string `protobuf:"bytes,17,rep,name=update_guard_plugin_ids,json=updateGuardPluginIds,proto3" json:"updateGuardPluginIds,omitempty"`
	// HostStorageId selects the Storage ID on the plugin host bus that plugin
	// instance volumes allocate through. Empty keeps the host default storage.
	// Top-level plugins leave this unset; instance-scoped runtimes set it to
	// route the instance's volumes through host-owned storage.
	HostStorageId string `protobuf:"bytes,18,opt,name=host_storage_id,json=hostStorageId,proto3" json:"hostStorageId,omitempty"`
	// ExternalPluginIds lists plugins another controller on the bus supplies.
	// The scheduler leaves their LoadPlugin directives to that controller.
	ExternalPluginIds []string `protobuf:"bytes,19,rep,name=external_plugin_ids,json=externalPluginIds,proto3" json:"externalPluginIds,omitempty"`
	// contains filtered or unexported fields
}

Config is the plugin host scheduler controller configuration.

Manages available plugin hosts and running plugins. Manages downloading the most appropriate manifest for the available plugin hosts. Manages executing plugins via the plugin hosts.

func NewConfig 

func NewConfig(
	instanceKey,
	engineID,
	objectKey,
	volumeID,
	peerID string,
	watchFetchManifest,
	disableStoreManifest,
	disableCopyManifest bool,
	noCopyBucketIDs ...string,
) *Config

NewConfig constructs a new controller config. Sets the most important fields only.

func (*Config) BuildExecBackoff 

func (c *Config) BuildExecBackoff() *backoff.Backoff

BuildExecBackoff gets the ExecBackoff and fills defaults if applicable.

func (*Config) BuildFetchBackoff 

func (c *Config) BuildFetchBackoff() *backoff.Backoff

BuildFetchBackoff gets the FetchBackoff and fills defaults if applicable.

func (*Config) BuildStartupWaitBudget added in v0.57.2 

func (c *Config) BuildStartupWaitBudget() (time.Duration, error)

BuildStartupWaitBudget gets the StartupWaitBudgetDur and fills the default if unset. Rejects negative and unparseable durations.

func (*Config) CloneMessageVT 

func (m *Config) CloneMessageVT() protobuf_go_lite.CloneMessage

func (*Config) CloneVT 

func (m *Config) CloneVT() *Config

func (*Config) EqualMessageVT 

func (this *Config) EqualMessageVT(thatMsg any) bool

func (*Config) EqualVT 

func (this *Config) EqualVT(that *Config) bool

func (*Config) EqualsConfig 

func (c *Config) EqualsConfig(other config.Config) bool

EqualsConfig checks if the config is equal to another.

func (*Config) FilterPluginPlatformIDs added in v0.51.7 

func (c *Config) FilterPluginPlatformIDs(pluginID string, platformIDs []string) []string

FilterPluginPlatformIDs filters platform IDs through PlatformSelectionPolicies.

func (*Config) GetConfigID 

func (c *Config) GetConfigID() string

GetConfigID returns the unique string for this configuration type.

func (*Config) GetDisableCopyManifest 

func (x *Config) GetDisableCopyManifest() bool

func (*Config) GetDisableStoreManifest 

func (x *Config) GetDisableStoreManifest() bool

func (*Config) GetEngineId 

func (x *Config) GetEngineId() string

func (*Config) GetExecBackoff 

func (x *Config) GetExecBackoff() *backoff.Backoff

func (*Config) GetExternalPluginIds 

func (x *Config) GetExternalPluginIds() []string

func (*Config) GetFetchBackoff 

func (x *Config) GetFetchBackoff() *backoff.Backoff

func (*Config) GetFetchConcurrency 

func (x *Config) GetFetchConcurrency() uint32

func (*Config) GetHostStorageId 

func (x *Config) GetHostStorageId() string

func (*Config) GetInstanceKey added in v0.57.2 

func (x *Config) GetInstanceKey() string

func (*Config) GetMaterializerPluginId added in v0.57.4 

func (x *Config) GetMaterializerPluginId() string

func (*Config) GetNoCopyBucketIds added in v0.57.0 

func (x *Config) GetNoCopyBucketIds() []string

func (*Config) GetObjectKey 

func (x *Config) GetObjectKey() string

func (*Config) GetPeerId 

func (x *Config) GetPeerId() string

func (*Config) GetPlatformSelectionPolicies added in v0.51.7 

func (x *Config) GetPlatformSelectionPolicies() []*PlatformSelectionPolicy

func (*Config) GetStartupWaitBudgetDur added in v0.57.2 

func (x *Config) GetStartupWaitBudgetDur() string

func (*Config) GetUpdateGuardPluginIds 

func (x *Config) GetUpdateGuardPluginIds() []string

func (*Config) GetVerbose 

func (x *Config) GetVerbose() bool

func (*Config) GetVolumeId 

func (x *Config) GetVolumeId() string

func (*Config) GetWatchFetchManifest 

func (x *Config) GetWatchFetchManifest() bool

func (*Config) MarshalJSON 

func (x *Config) MarshalJSON() ([]byte, error)

MarshalJSON marshals the Config to JSON.

func (*Config) MarshalProtoJSON 

func (x *Config) MarshalProtoJSON(s *json.MarshalState)

MarshalProtoJSON marshals the Config message to JSON.

func (*Config) MarshalProtoText 

func (x *Config) MarshalProtoText() string

func (*Config) MarshalToSizedBufferVT 

func (m *Config) MarshalToSizedBufferVT(dAtA []byte) (int, error)

func (*Config) MarshalToVT 

func (m *Config) MarshalToVT(dAtA []byte) (int, error)

func (*Config) MarshalVT 

func (m *Config) MarshalVT() (dAtA []byte, err error)

func (*Config) ParsePeerID 

func (c *Config) ParsePeerID() (peer.ID, error)

ParsePeerID parses the peer ID field.

func (*Config) ProtoMessage 

func (*Config) ProtoMessage()

func (*Config) Reset 

func (x *Config) Reset()

func (*Config) SizeVT 

func (m *Config) SizeVT() (n int)

func (*Config) String 

func (x *Config) String() string

func (*Config) UnmarshalJSON 

func (x *Config) UnmarshalJSON(b []byte) error

UnmarshalJSON unmarshals the Config from JSON.

func (*Config) UnmarshalProtoJSON 

func (x *Config) UnmarshalProtoJSON(s *json.UnmarshalState)

UnmarshalProtoJSON unmarshals the Config message from JSON.

func (*Config) UnmarshalVT 

func (m *Config) UnmarshalVT(dAtA []byte) error

func (*Config) Validate 

func (c *Config) Validate() error

Validate validates the configuration. This is a cursory validation to see if the values "look correct."

type Controller 

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

Controller implements the plugin host scheduler controller.

Manages available plugin hosts and running plugins. Manages downloading the most appropriate manifest for the available plugin hosts. Manages executing plugins via the plugin hosts.

Only one of these controllers should be running.

func FindControllerOnBus added in v0.51.7 

func FindControllerOnBus(b bus.Bus) *Controller

FindControllerOnBus returns the first plugin host scheduler on b.

func NewController 

func NewController(
	le *logrus.Entry,
	bus bus.Bus,
	conf *Config,
) *Controller

NewController constructs a new controller.

func WaitControllerOnBus added in v0.57.0 

func WaitControllerOnBus(ctx context.Context, b bus.Bus) (*Controller, error)

WaitControllerOnBus waits for a plugin host scheduler to be added to b.

func (*Controller) AddPinnedPluginReference 

func (c *Controller) AddPinnedPluginReference(pluginID, instanceKey, manifestRoot string) (bldr_plugin.RunningPluginRef, func())

AddPinnedPluginReference retains an exact manifest in a separate host worker.

func (*Controller) AddPluginReference 

func (c *Controller) AddPluginReference(pluginID, instanceKey string) (bldr_plugin.RunningPluginRef, func())

AddPluginReference adds a reference to the plugin, returning the RunningPlugin handle and a release function. instanceKey may be empty for shared (non-instanced) plugins.

func (*Controller) AddSelectedPluginReference 

func (c *Controller) AddSelectedPluginReference(pluginID, instanceKey string, selected ...*bldr_manifest.ManifestRef) (bldr_plugin.RunningPluginRef, func())

AddSelectedPluginReference retains a desired artifact without replacing the binding. The newest retained selection wins; removing it restores the preceding demand.

func (*Controller) Close 

func (c *Controller) Close() error

Close releases any resources used by the controller. Error indicates any issue encountered releasing.

func (*Controller) Execute 

func (c *Controller) Execute(rctx context.Context) (rerr error)

Execute executes the controller. Returning nil ends execution. Returning an error triggers a retry with backoff.

func (*Controller) GetControllerInfo 

func (c *Controller) GetControllerInfo() *controller.Info

GetControllerInfo returns information about the controller.

func (*Controller) GetInstanceKey 

func (c *Controller) GetInstanceKey() string

GetInstanceKey returns the instance key the scheduler was configured with.

func (*Controller) GetPluginStatusCtr added in v0.51.0 

func (c *Controller) GetPluginStatusCtr() ccontainer.Watchable[*bldr_plugin.PluginStatusSnapshot]

GetPluginStatusCtr returns the scheduler's live plugin-status snapshot.

func (*Controller) HandleDirective 

func (c *Controller) HandleDirective(
	ctx context.Context,
	inst directive.Instance,
) ([]directive.Resolver, error)

HandleDirective asks if the handler can resolve the directive. If it can, it returns a resolver. If not, returns nil. Any unexpected errors are returned for logging. It is safe to add a reference to the directive during this call. The context tasked is canceled when the directive instance expires.

func (*Controller) IsPluginRunning added in v0.51.5 

func (c *Controller) IsPluginRunning(pluginID string) bool

IsPluginRunning reports whether any instance for pluginID is running.

func (*Controller) SetManifestCopyGate added in v0.57.0 

func (c *Controller) SetManifestCopyGate(gate ManifestCopyGate)

SetManifestCopyGate sets the runtime readiness gate for startup manifest copies. A nil gate preserves immediate copy behavior.

func (*Controller) WaitPluginClient 

func (c *Controller) WaitPluginClient(ctx context.Context, released func(), pluginID string) (srpc.Client, func(), error)

WaitPluginClient waits for an RPC client for a plugin.

if pluginID is invalid, returns an error.

Released is a function to call if the client becomes invalid. Returns nil, nil, err if any error. Returns nil, nil, nil to skip resolving the client. Otherwise returns client, releaseFunc, nil

func (*Controller) WaitPluginHostClient 

func (c *Controller) WaitPluginHostClient(ctx context.Context, released func()) (srpc.Client, func(), error)

WaitPluginHostClient waits for an RPC client for the plugin host.

Released is a function to call if the client becomes invalid. Returns nil, nil, err if any error. Returns nil, nil, nil to skip resolving the client. Otherwise returns client, releaseFunc, nil

func (*Controller) WaitPluginsRunning added in v0.51.7 

func (c *Controller) WaitPluginsRunning(ctx context.Context, pluginIDs []string) error

WaitPluginsRunning waits until all shared instances in pluginIDs report running, or returns the first scheduler error recorded for one of them.

type Factory 

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

Factory constructs the controller factory

func NewFactory 

func NewFactory(bus bus.Bus) *Factory

NewFactory builds the factory.

func (*Factory) Construct 

func (t *Factory) Construct(
	ctx context.Context,
	conf config.Config,
	opts controller.ConstructOpts,
) (controller.Controller, error)

Construct constructs the associated controller given configuration.

func (*Factory) ConstructConfig 

func (t *Factory) ConstructConfig() config.Config

ConstructConfig constructs an instance of the controller configuration.

func (*Factory) GetConfigID 

func (t *Factory) GetConfigID() string

GetConfigID returns the configuration ID for the controller.

func (*Factory) GetControllerID 

func (t *Factory) GetControllerID() string

GetControllerID returns the unique ID for the controller.

func (*Factory) GetVersion 

func (t *Factory) GetVersion() controller.Version

GetVersion returns the version of this controller.

type ManifestCopyGate added in v0.57.0 

type ManifestCopyGate interface {
	// IsReady reports whether new copies can start without waiting.
	IsReady() bool
	// WaitReady waits until copies can start or the context is canceled.
	WaitReady(ctx context.Context) error
}

ManifestCopyGate delays startup manifest copies until its readiness boundary.

type PlatformSelectionPolicy added in v0.51.7 

type PlatformSelectionPolicy struct {

	// PlatformId is the plugin host platform ID to restrict, such as
	// "web/js/wasm".
	PlatformId string `protobuf:"bytes,1,opt,name=platform_id,json=platformId,proto3" json:"platformId,omitempty"`
	// AllowedPluginIds are the only plugin IDs that may select PlatformId when
	// nonempty.
	AllowedPluginIds []string `protobuf:"bytes,2,rep,name=allowed_plugin_ids,json=allowedPluginIds,proto3" json:"allowedPluginIds,omitempty"`
	// DeniedPluginIds are the plugin IDs that cannot select PlatformId. This
	// leaves unlisted and externally supplied plugins eligible.
	DeniedPluginIds []string `protobuf:"bytes,3,rep,name=denied_plugin_ids,json=deniedPluginIds,proto3" json:"deniedPluginIds,omitempty"`
	// contains filtered or unexported fields
}

PlatformSelectionPolicy restricts one plugin host platform by plugin ID.

func (*PlatformSelectionPolicy) CloneMessageVT added in v0.51.7 

func (m *PlatformSelectionPolicy) CloneMessageVT() protobuf_go_lite.CloneMessage

func (*PlatformSelectionPolicy) CloneVT added in v0.51.7 

func (m *PlatformSelectionPolicy) CloneVT() *PlatformSelectionPolicy

func (*PlatformSelectionPolicy) EqualMessageVT added in v0.51.7 

func (this *PlatformSelectionPolicy) EqualMessageVT(thatMsg any) bool

func (*PlatformSelectionPolicy) EqualVT added in v0.51.7 

func (this *PlatformSelectionPolicy) EqualVT(that *PlatformSelectionPolicy) bool

func (*PlatformSelectionPolicy) GetAllowedPluginIds added in v0.51.7 

func (x *PlatformSelectionPolicy) GetAllowedPluginIds() []string

func (*PlatformSelectionPolicy) GetDeniedPluginIds 

func (x *PlatformSelectionPolicy) GetDeniedPluginIds() []string

func (*PlatformSelectionPolicy) GetPlatformId added in v0.51.7 

func (x *PlatformSelectionPolicy) GetPlatformId() string

func (*PlatformSelectionPolicy) MarshalJSON added in v0.51.7 

func (x *PlatformSelectionPolicy) MarshalJSON() ([]byte, error)

MarshalJSON marshals the PlatformSelectionPolicy to JSON.

func (*PlatformSelectionPolicy) MarshalProtoJSON added in v0.51.7 

func (x *PlatformSelectionPolicy) MarshalProtoJSON(s *json.MarshalState)

MarshalProtoJSON marshals the PlatformSelectionPolicy message to JSON.

func (*PlatformSelectionPolicy) MarshalProtoText added in v0.51.7 

func (x *PlatformSelectionPolicy) MarshalProtoText() string

func (*PlatformSelectionPolicy) MarshalToSizedBufferVT added in v0.51.7 

func (m *PlatformSelectionPolicy) MarshalToSizedBufferVT(dAtA []byte) (int, error)

func (*PlatformSelectionPolicy) MarshalToVT added in v0.51.7 

func (m *PlatformSelectionPolicy) MarshalToVT(dAtA []byte) (int, error)

func (*PlatformSelectionPolicy) MarshalVT added in v0.51.7 

func (m *PlatformSelectionPolicy) MarshalVT() (dAtA []byte, err error)

func (*PlatformSelectionPolicy) ProtoMessage added in v0.51.7 

func (*PlatformSelectionPolicy) ProtoMessage()

func (*PlatformSelectionPolicy) Reset added in v0.51.7 

func (x *PlatformSelectionPolicy) Reset()

func (*PlatformSelectionPolicy) SizeVT added in v0.51.7 

func (m *PlatformSelectionPolicy) SizeVT() (n int)

func (*PlatformSelectionPolicy) String added in v0.51.7 

func (x *PlatformSelectionPolicy) String() string

func (*PlatformSelectionPolicy) UnmarshalJSON added in v0.51.7 

func (x *PlatformSelectionPolicy) UnmarshalJSON(b []byte) error

UnmarshalJSON unmarshals the PlatformSelectionPolicy from JSON.

func (*PlatformSelectionPolicy) UnmarshalProtoJSON added in v0.51.7 

func (x *PlatformSelectionPolicy) UnmarshalProtoJSON(s *json.UnmarshalState)

UnmarshalProtoJSON unmarshals the PlatformSelectionPolicy message from JSON.

func (*PlatformSelectionPolicy) UnmarshalVT added in v0.51.7 

func (m *PlatformSelectionPolicy) UnmarshalVT(dAtA []byte) error

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.