workers

package module
v0.0.0-...-1e91397 Latest Latest
Warning

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

 Go to latest Published: Jul 11, 2026 License: Apache-2.0 Imports: 6 Imported by: 0

README

workers

Bounded, composable worker primitives for the Wago WebAssembly runtime.

CI Coverage

Table of Contents

Overview

workers is an optional Wago plugin that turns a WebAssembly table entry into a bounded, host-supervised worker. Each worker is a forked instance of the caller, driven on its own goroutine, fed by a fixed-size, byte-bounded mailbox. The host observes every message and every exit.

Workers does not define PIDs, guest-visible mailboxes, monitors, supervision trees, restart policy, or a guest ABI. Those are policies you build on top of the primitives here: Spawn, Send, DispatchNext, Link, and Kill, using the host callbacks OnMessage and OnExit.

What you get out of the box:

  • Bounded by construction: every worker has a capped queue depth, a max payload size, and a max total queued bytes. Send copies the payload and never blocks.
  • Host-supervised: OnMessage sees every delivered message; OnExit sees every return, failure, or kill, with the cause.
  • Lifecycle-safe: a linked worker is torn down cooperatively when its creating instance closes, and every worker is drained when the runtime shuts down.

Stability: experimental (v0.1.0). The API may change before v1.0.0.

Installation

If you have the wago CLI installed:

wago pkg install github.com/wago-org/workers

or use go get:

go get github.com/wago-org/workers

The plugin requires two privileged capabilities - instance.manage (to fork and manage worker instances) and instance.lifecycle (to observe parent shutdown). Grant them in your wago.json:

{
  "dependencies": ["github.com/wago-org/workers"],
  "plugins": [
    {
      "name": "workers",
      "capabilities": {
        "instance.manage": { "maxInstances": 64 },
        "instance.lifecycle": true
      }
    }
  ]
}

When you register the plugin programmatically, grant the same capabilities with wago.WithPluginGrants (see Usage). Grants are enforced in strict mode; omitting the option registers the plugin without capability checks.

Concepts

Term Meaning
Plugin workers.New() - the wago.Extension you register with a runtime. It provides one service.
Service (*workers.Workers) The host-side handle you use to spawn, send, link, and kill workers, and to register OnMessage / OnExit observers. Obtain it with plugin.Service() or look it up through workers.ServiceKey.
Worker A forked child instance running one table entry with the exact Wasm signature () -> (), on its own goroutine. Identified by a WorkerID.
Mailbox A fixed-capacity, byte-bounded FIFO queue owned by each worker. Send enqueues; the worker drains it cooperatively via DispatchNext.
Creator The instance that spawned the worker. Link ties a worker's lifetime to its exact creator.

A worker's life: spawn → run its table entry → receive messages cooperatively → exit (return, fail, or kill) → OnExit fires → resources freed.

Usage

Register the plugin, grant its capabilities, and attach host observers:

package main

import (
	"log"

	"github.com/wago-org/wago"
	"github.com/wago-org/workers"
)

func main() {
	workerPlugin := workers.New()

	rt := wago.NewRuntime()
	if err := rt.Use(workerPlugin, wago.WithPluginGrants(
		wago.PluginManagedInstances, // "instance.manage"
		wago.PluginInstanceHooks,    // "instance.lifecycle"
	)); err != nil {
		log.Fatal(err)
	}
	defer rt.Close()

	service := workerPlugin.Service()

	// Observe every message delivered to any worker.
	service.OnMessage(func(ctx *workers.MessageContext) error {
		// Decode ctx.Tag / ctx.Payload, or write into ctx.Caller.Memory().
		// Returning an error stops that worker with WorkerFailed.
		return nil
	})

	// Observe every worker exit: return, failure, or kill.
	service.OnExit(func(ctx *workers.WorkerExitContext) {
		// Build monitoring, restarts, signalling, or logging here.
		log.Printf("worker %d exited: kind=%d err=%v", ctx.WorkerID, ctx.Kind, ctx.Err)
	})

	// ... compile and instantiate guest modules that spawn/drive workers.
}

Service() returns the same handle that a downstream plugin can retrieve through the exported workers.ServiceKey, so composing plugins share one worker registry. Inside your plugin's Register, take a typed reference and resolve it once the runtime is wired up:

ref, err := plugin.Require(reg, workers.ServiceKey)
if err != nil {
	return err
}
// later, when handling a call:
svc, err := ref.Get()

API

All host-side operations that cross into a guest - Spawn, DispatchNext, Current, Link - must be called from inside a synchronous host import, where caller is the active wago.HostModule for that call. Send and Kill take a WorkerID and may be called from anywhere.

Spawning a worker

Spawn forks the calling instance and runs the table entry at tableIndex. The entry must have the exact Wasm signature () -> (); otherwise Spawn returns a validation error and the child is discarded.

id, err := service.Spawn(caller, tableIndex, workers.WorkerOptions{
	QueueCapacity:   64,
	MaxPayloadBytes: 64 << 10, // 64 KiB
	MaxQueueBytes:   1 << 20,  // 1 MiB
})
if err != nil {
	return err
}

Zero-valued options fall back to the package defaults (see Worker options and limits).

Sending messages

Send copies the payload and returns immediately - it never blocks the caller. It fails if the worker is unknown, stopping, or its mailbox is at capacity (by count or by bytes), or if the payload exceeds the worker's MaxPayloadBytes.

if err := service.Send(id, 42 /* tag */, []byte("hello")); err != nil {
	// e.g. errors.Is(err, workers.ErrWorkerQueueFull)
	return err
}

Because the payload is copied, the caller may reuse or mutate its buffer as soon as Send returns.

Receiving inside a worker

A worker drains one message per DispatchNext, which you expose to the guest through a plugin-defined host import. It delivers the next queued message to every registered OnMessage observer, blocking cooperatively until a message arrives or the worker stops.

// Inside the host import the guest calls to receive:
if err := service.DispatchNext(caller); err != nil {
	return pluginErrno(err) // map to a guest-visible errno
}

A worker can learn its own ID with Current:

id, err := service.Current(caller)

Linking and termination

Link ties a worker's lifetime to its exact creator: when the creating instance closes, the runtime's BeforeClose hook cooperatively stops every linked worker and waits for them to drain. A worker cannot link to itself, and only the creator may link it.

if err := service.Link(caller, childID); err != nil {
	return err
}

Kill requests cooperative termination of any worker by ID. The worker's in-flight DispatchNext unblocks, its queue is cleared, and it exits with WorkerKilled.

_ = service.Kill(id)

Every exit - whether the table entry returned, an OnMessage observer returned an error, or the worker was killed - surfaces through OnExit:

service.OnExit(func(ctx *workers.WorkerExitContext) {
	switch ctx.Kind {
	case workers.WorkerReturned: // clean return from the table entry
	case workers.WorkerFailed:   // trap or observer error; see ctx.Err
	case workers.WorkerKilled:   // Kill, parent close, or runtime shutdown
	}
})

Worker options and limits

WorkerOptions bounds a single worker. A zero field takes the package default. Values above the hard maximum (or a MaxQueueBytes smaller than MaxPayloadBytes) are rejected with ErrInvalidWorkerOptions.

Field Default Maximum Meaning
QueueCapacity 64 65536 Max number of queued messages.
MaxPayloadBytes 64 KiB 16 MiB Max size of a single Send payload.
MaxQueueBytes 1 MiB 64 MiB Max total bytes queued at once (must be ≥ MaxPayloadBytes).

The corresponding exported constants are DefaultWorkerQueueCapacity, DefaultWorkerMaxPayloadBytes, DefaultWorkerMaxQueueBytes, MaxWorkerQueueCapacity, MaxWorkerPayloadBytes, and MaxWorkerQueueBytes.

WorkerOptions bounds one worker; WorkerLimits bounds the whole service, so a guest cannot exhaust the host by spawning workers without end. It always applies — even when the host grants instance.manage with no maxInstances budget — and complements that core budget rather than replacing it. Pass it at construction:

workerPlugin := workers.New(workers.WithLimits(workers.WorkerLimits{
	MaxLiveWorkers: 128,       // default 64
	MaxQueueBytes:  128 << 20, // default 64 MiB, summed across live workers
}))
Field Default Meaning
MaxLiveWorkers 64 Max workers live at once. Spawn returns ErrWorkerQuotaExceeded past it.
MaxQueueBytes 64 MiB Max total per-worker MaxQueueBytes reservation summed across live workers.

Errors

All errors are comparable sentinels - match them with errors.Is.

Error When
ErrWorkersInactive Service is nil or not registered.
ErrInvalidWorkerOptions Options exceed a hard limit or are inconsistent.
ErrInvalidWorkerCaller Operation needs the current plugin host caller and none is active.
ErrWorkerImportLifetime The worker would inherit a borrowed import it cannot safely keep.
ErrWorkerNotFound No worker with that ID (or not owned by the caller).
ErrWorkerStopping The worker is already shutting down.
ErrWorkerQueueFull Mailbox at capacity by count or bytes.
ErrWorkerDispatchActive DispatchNext is already running for that worker.
ErrPayloadTooLarge Payload exceeds the worker's MaxPayloadBytes.
ErrWorkerIDExhausted The WorkerID space is exhausted.
ErrInvalidWorkerLink Link target is not the caller's direct child, or is self.
ErrWorkerKilled Exit cause: Kill was requested.
ErrWorkerParentClosed Exit cause: the linked creator instance closed.
ErrWorkerRuntimeClosed Exit cause: the runtime shut down.
ErrWorkerQuotaExceeded Spawn would exceed the service-wide WorkerLimits.

Examples

The end-to-end integration test is the canonical, runnable example: it defines a guest module that imports spawn/next, spawns a worker, and verifies the host copies a message and observes a clean exit. See workers_test.go - in particular TestPluginSpawnsCopiesMessageAndStops.

A generated Wago host does not import this package directly; it blank-imports the register subpackage, which activates the plugin's init-time registration:

import _ "github.com/wago-org/workers/register"

Testing

go test ./...

The suite uses github.com/wago-org/wago/testutil/wasmtest to hand-build guest modules, so no external toolchain is required. Run with the race detector while iterating on the concurrency paths:

go test -race ./...

Architecture

  • workers.go - the whole plugin: the Plugin extension, the Workers service, the per-worker mailbox and goroutine, and the lifecycle wiring (BeforeClose teardown, runtime-close drain).
  • register/ - a blank-import shim that activates init-time registration for Wago-generated hosts.
  • wago.json - the package manifest declaring the plugin and its required capabilities.

Design notes:

  • Each worker runs on one goroutine (worker.run) that invokes the table entry and, on return, resolves the exit kind, closes the child instance, unregisters it, and fans out to OnExit observers (each guarded against panics).
  • The mailbox is a fixed-size ring buffer guarded by a mutex; Send enqueues and signals, DispatchNext dequeues and delivers. Back-pressure is explicit: a full queue rejects rather than blocks.
  • Spawn forks through the managed-instance API and validates the target table entry before the worker is made visible, so a bad spawn leaves no partial state.

Contributing

Contributions are welcome! Please:

  • Run go test -race ./... and go vet ./... before opening a pull request.
  • Keep the plugin's boundaries intact - no PIDs, guest mailboxes, supervision, or guest ABI belong in this package; those are policies for layers built on top.
  • Follow standard Go formatting (gofmt) and conventional commit messages.

License

This project is distributed under the Apache License 2.0. Work on this project is done out of passion - if you want to support it financially, you can donate through GitHub Sponsors.

Contact

Please file issues at GitHub Issues. To chat, join the Wago Discord.

Documentation

Overview

Package workers provides bounded WebAssembly workers as an optional Wago plugin.

Index

Constants

View Source 
const (
	DefaultWorkerQueueCapacity   uint32 = 64
	DefaultWorkerMaxPayloadBytes uint32 = 64 << 10
	DefaultWorkerMaxQueueBytes   uint32 = 1 << 20
	MaxWorkerQueueCapacity       uint32 = 1 << 16
	MaxWorkerPayloadBytes        uint32 = 16 << 20
	MaxWorkerQueueBytes          uint32 = 64 << 20

	// DefaultMaxLiveWorkers bounds the number of simultaneously live workers for
	// one service when WorkerLimits.MaxLiveWorkers is left zero. Each worker owns a
	// managed instance, a goroutine, and a foreign stack, so an unbounded count is a
	// denial-of-service vector; a default keeps Spawn bounded even when the host
	// grants instance.manage without a maxInstances budget.
	DefaultMaxLiveWorkers uint32 = 64
	// DefaultMaxServiceQueueBytes bounds the total queued-payload reservation across
	// all live workers when WorkerLimits.MaxQueueBytes is left zero.
	DefaultMaxServiceQueueBytes uint64 = 64 << 20
)
View Source 
const (
	ErrWorkersInactive      workerError = "worker service is not active"
	ErrInvalidWorkerOptions workerError = "invalid worker options"
	ErrInvalidWorkerCaller  workerError = "worker operation requires a current plugin host caller"
	ErrWorkerImportLifetime workerError = "worker cannot safely inherit a borrowed import"
	ErrWorkerNotFound       workerError = "worker not found"
	ErrWorkerStopping       workerError = "worker is stopping"
	ErrWorkerQueueFull      workerError = "worker queue is full"
	ErrWorkerDispatchActive workerError = "worker message dispatch is already active"
	ErrPayloadTooLarge      workerError = "worker payload is too large"
	ErrWorkerIDExhausted    workerError = "worker ID space exhausted"
	ErrInvalidWorkerLink    workerError = "invalid worker link"
	ErrWorkerKilled         workerError = "worker killed"
	ErrWorkerParentClosed   workerError = "worker parent closed"
	ErrWorkerRuntimeClosed  workerError = "worker runtime closed"
	ErrWorkerQuotaExceeded  workerError = "worker service resource quota exceeded"
)
View Source 
const PluginName = "workers"

Variables

View Source 
var ServiceKey = plugin.NewServiceKey[*Workers]("wago.workers/v1")

Functions

This section is empty.

Types

type MessageContext 

type MessageContext struct {
	WorkerID WorkerID
	Tag      uint64
	Payload  []byte
	Caller   wago.HostModule
}

type Option 

type Option func(*Plugin)

Option configures the workers Plugin at construction.

func WithLimits 

func WithLimits(l WorkerLimits) Option

WithLimits sets the aggregate resource limits for the worker service. Zero fields fall back to the package defaults (see WorkerLimits).

type Plugin 

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

func New 

func New(opts ...Option) *Plugin

New creates the workers plugin. Pass WithLimits to override the default aggregate resource caps.

func (*Plugin) Info 

func (*Plugin) Info() wago.ExtensionInfo

func (*Plugin) Register 

func (p *Plugin) Register(reg *wago.Registry) error

func (*Plugin) Service 

func (p *Plugin) Service() *Workers

func (*Plugin) Stop 

func (p *Plugin) Stop(context.Context) error

type WorkerExitContext 

type WorkerExitContext struct {
	WorkerID WorkerID
	Kind     WorkerExitKind
	Err      error
}

type WorkerExitKind 

type WorkerExitKind uint8
const (
	WorkerReturned WorkerExitKind = iota + 1
	WorkerFailed
	WorkerKilled
)

type WorkerID 

type WorkerID uint64

type WorkerLimits 

type WorkerLimits struct {
	// MaxLiveWorkers is the maximum number of simultaneously live workers.
	MaxLiveWorkers uint32
	// MaxQueueBytes is the maximum total per-worker queue-byte reservation summed
	// across all live workers (each worker reserves its MaxQueueBytes for its
	// lifetime).
	MaxQueueBytes uint64
}

WorkerLimits bounds the aggregate resources one worker service may hold at once, independent of the per-worker WorkerOptions. Zero fields take the package defaults. It complements — and does not replace — the core instance.manage maxInstances budget: this cap always applies, so workers stay bounded even when the host grants instance.manage without a budget.

type WorkerOptions 

type WorkerOptions struct {
	QueueCapacity   uint32
	MaxPayloadBytes uint32
	MaxQueueBytes   uint32
}

type Workers 

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

func (*Workers) Current 

func (w *Workers) Current(caller wago.HostModule) (WorkerID, error)

func (*Workers) DispatchNext 

func (w *Workers) DispatchNext(caller wago.HostModule) error

func (*Workers) Kill 

func (w *Workers) Kill(id WorkerID) error
func (w *Workers) Link(caller wago.HostModule, childID WorkerID) error

func (*Workers) OnExit 

func (w *Workers) OnExit(fns ...func(*WorkerExitContext))

func (*Workers) OnMessage 

func (w *Workers) OnMessage(fns ...func(*MessageContext) error)

func (*Workers) Send 

func (w *Workers) Send(id WorkerID, tag uint64, payload []byte) error

func (*Workers) Spawn 

func (w *Workers) Spawn(caller wago.HostModule, tableIndex uint32, opts WorkerOptions) (WorkerID, error)

Source Files

View all Source files

Directories

Path Synopsis
Package register activates the workers plugin's init-time registration for Wago-generated hosts.
 Package register activates the workers plugin's init-time registration for Wago-generated hosts.

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.