Building a Host

For developers putting Dwarf behind a real application. It turns Access Control Is the Host’s Job into code. Read Getting Started first for the two-method Host interface.

The engine has no notion of a caller: no authentication, no authorization, no tenants. Every access decision belongs to the layer you build around it. This page builds that layer for a multi-tenant service, as one type, App, that is three things at once:

the front door your API handlers call to start and manage flows;
the engine’s host, which checks the caller’s identity before loading a graph or running a task;
the record of ownership that says which tenant created which flow.

Local dispatch, from a URL to a graph or a Go function, is delegated to hosts.Mux, so App only adds the access checks.

The Pieces

// Claims identify the caller. In a real service they come from a verified token.
type Claims struct {
	Subject string `json:"sub"`
	Tenant  string `json:"tenant"`
}

var (
	ErrQuotaExceeded = errors.New("quota exceeded")
	ErrForbidden     = errors.New("forbidden")
)

// validTenant is the shape of a fairness key: 1 to 256 ASCII letters, digits, '-', '_' and '.'.
var validTenant = regexp.MustCompile(`^[A-Za-z0-9._-]{1,256}$`)

// tenantOf returns the caller's tenant, refusing a caller without one, or with one that is
// not a valid fairness key.
func tenantOf(caller Claims) (string, error) {
	if !validTenant.MatchString(caller.Tenant) {
		return "", ErrForbidden
	}
	return caller.Tenant, nil
}

// App wraps the engine: it is the authenticated front door, the engine's Host,
// and the record of which tenant owns each flow it created.
type App struct {
	eng   *engine.Engine
	mux   *hosts.Mux // local dispatch: graphs and task functions by URL
	quota int        // most running flows a tenant may have at once

	mu     sync.Mutex
	owners map[string]string // flow key -> tenant, for every key this app minted
}

func NewApp(eng *engine.Engine, mux *hosts.Mux, quota int) *App {
	return &App{eng: eng, mux: mux, quota: quota, owners: map[string]string{}}
}

hosts.Mux holds the graphs and task functions, registered with mux.HandleGraph and mux.HandleTask. The owners map stands in for your own storage: in production, record each flow key next to its tenant in your database.

The tenant becomes the flow’s fairness key, so it must be a valid one: at most 256 ASCII letters, digits, -, _ and ., compared case-sensitively. tenantOf refuses anything else, including an empty tenant, which matters twice: Create would reject an invalid key, and an empty Query.FairnessKey disables the filter, so a tenant-scoped list would return every tenant’s flows. If your tenant IDs can hold other characters (an email address, say), or differ only in case, map each to a key that fits, such as its database ID or a hex digest.

Starting a Flow: Quota, Claims and Tenant

The front door is the one place flows are created. It enforces the tenant’s quota, then creates the flow with the caller’s claims as baggage and the tenant as the fairness key:

// Start is the front door for creating a flow.
func (a *App) Start(ctx context.Context, caller Claims, workflowURL string, input map[string]any) (string, error) {
	tenant, err := tenantOf(caller)
	if err != nil {
		return "", err
	}
	// The engine bounds nothing, so enforce the tenant's quota before Create.
	open, err := a.eng.ListOpen(ctx, workflow.Query{
		FairnessKey: tenant,  // this tenant's flows only
		Limit:       a.quota, // enough to tell whether the quota is reached
	})
	if err != nil {
		return "", err
	}
	if len(open) >= a.quota {
		return "", ErrQuotaExceeded
	}
	flowKey, err := a.eng.Create(ctx, workflowURL, input, &workflow.FlowOptions{
		FairnessKey: tenant, // fair scheduling between tenants, and tenant-scoped lists
		Baggage:     caller, // the caller's claims, visible to LoadGraph and every task
	})
	if err != nil {
		return "", err
	}
	a.mu.Lock()
	a.owners[flowKey] = tenant
	a.mu.Unlock()
	return flowKey, nil
}

The quota comes first because the engine bounds nothing: not how many flows a tenant starts, not how large their input is. It counts the tenant’s open flows, which ListOpen returns: those running and those interrupted, waiting for input, since both are still in progress. The check and the Create are two calls, so concurrent requests from one tenant can pass the check together; serialize them per tenant if the quota must be exact.
The claims ride along as baggage. They are fixed for the flow’s life, inherited by its subgraphs, and visible to LoadGraph and to every task, so a task acts on behalf of the person who started the flow.
The tenant is the fairness key, which gives each tenant a fair share of the workers and makes a tenant-scoped list, and the quota check, a single query.

The Host: Checking Claims Before Delegating

The engine calls LoadGraph when a flow or subgraph is created, and ExecuteTask for every task, with the flow’s baggage on the context. App reads the claims with workflow.BaggageFrom and refuses anything without a tenant, then delegates to the mux:

// LoadGraph implements engine.Host. It sees the caller's claims as baggage, so it
// can refuse a workflow the tenant may not run.
func (a *App) LoadGraph(ctx context.Context, workflowURL string) (*workflow.Graph, error) {
	if workflow.BaggageFrom(ctx).GetString("tenant") == "" {
		return nil, ErrForbidden
	}
	return a.mux.LoadGraph(ctx, workflowURL)
}

// ExecuteTask implements engine.Host. Every task runs on behalf of the caller who
// started the flow, whose claims ride along as baggage.
func (a *App) ExecuteTask(ctx context.Context, taskURL string, f *workflow.Flow) error {
	if workflow.BaggageFrom(ctx).GetString("tenant") == "" {
		return ErrForbidden
	}
	return a.mux.ExecuteTask(ctx, taskURL, f)
}

This is where per-tenant rules belong: a workflow a tenant may not run, a task that needs a particular role. A task itself can read the same claims, for example to act on the caller’s behalf downstream: workflow.BaggageFrom(ctx).GetString("sub").

Managing a Flow: Ownership, Not Possession

A flow key is a capability, not a permission. Anyone holding one can resume, terminate, fork and read the flow through the engine, so the front door checks ownership against its own record before passing an operation on:

// owns reports whether the flow was created by this app for the caller's tenant.
// Holding a flow key is not enough: a key is a capability, not a permission.
func (a *App) owns(caller Claims, flowKey string) bool {
	tenant, err := tenantOf(caller)
	if err != nil {
		return false
	}
	a.mu.Lock()
	defer a.mu.Unlock()
	return a.owners[flowKey] == tenant
}

// Resume continues an interrupted flow, if the caller's tenant owns it.
func (a *App) Resume(ctx context.Context, caller Claims, flowKey string, data any) error {
	if !a.owns(caller, flowKey) {
		return ErrForbidden
	}
	return a.eng.Resume(ctx, flowKey, data)
}

// Terminate stops a flow, if the caller's tenant owns it.
func (a *App) Terminate(ctx context.Context, caller Claims, flowKey, reason string) error {
	if !a.owns(caller, flowKey) {
		return ErrForbidden
	}
	return a.eng.Terminate(ctx, flowKey, reason)
}

// ListOpen returns the caller's tenant's open flows, and nobody else's.
func (a *App) ListOpen(ctx context.Context, caller Claims, q workflow.Query) ([]workflow.FlowSummary, error) {
	tenant, err := tenantOf(caller)
	if err != nil {
		return nil, err // an empty key would list every tenant's flows
	}
	q.FairnessKey = tenant
	return a.eng.ListOpen(ctx, q)
}

// ListClosed returns the caller's tenant's closed flows, and nobody else's.
func (a *App) ListClosed(ctx context.Context, caller Claims, q workflow.Query) ([]workflow.FlowSummary, error) {
	tenant, err := tenantOf(caller)
	if err != nil {
		return nil, err
	}
	q.FairnessKey = tenant
	return a.eng.ListClosed(ctx, q)
}

The lists are the operations that hand out keys wholesale, so the front door always scopes them to the caller’s tenant by overriding Query.FairnessKey. A flow that closes leaves ListOpen at once and appears in ListClosed about a minute later, once it is archived. Expose Fork, Snapshot, History, Step and Delete the same way: check ownership, then delegate; and scope Purge by overriding its Query.FairnessKey with tenantOf, as the lists do, since an empty key would purge every tenant’s flows. Keep the engine’s own methods off any interface a caller can reach.

Wiring It Up

mux := hosts.NewMux()
mux.HandleGraph("approval", approvalGraph)
mux.HandleTask("approval/request", requestApproval)
mux.HandleTask("approval/apply", applyChange)

eng := dwarf.NewEngine()
eng.SetShard(engine.ShardSpec{Index: 1, DSN: "postgres://user:pass@db:5432/dwarf"})
app := NewApp(eng, mux, 10) // up to 10 flows in progress per tenant
eng.SetHost(app)
if err := eng.Startup(ctx); err != nil {
	log.Fatal(err)
}
defer eng.Shutdown(ctx)

With this in place, a second tenant holding a leaked flow key gets ErrForbidden from Resume, and its listing never contains another tenant’s flows. To reach tasks over HTTP or in other languages, replace the mux with your own dispatch; see Tasks in Other Languages.

Next: Tasks in Other Languages.