flezha

package module
v0.1.1 Latest Latest
Warning

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

 Go to latest Published: Nov 4, 2026 License: Apache-2.0 Imports: 26 Imported by: 0

README

Flezha Go SDK

The Go SDK for Flezha: relays written as Go types, served by a runner that links flz_core, the runner kernel the Rust, Java, Node, Python, Ruby, PHP and Elixir SDKs also drive, into one executable.

Support tier: tier 2: it passes the full conformance run exactly as tier 1 does, with every capability present, and is permanently supported; it is not yet run by the maintainer in a live deployment.

Install

go get github.com/flezha/flezha_go_sdk
eval "$(go run github.com/flezha/flezha_go_sdk/cmd/flz-native@v0.1.1)"

The second line downloads the prebuilt core library for your platform (Linux or macOS, x64 or arm64), checks it against the release's checksums, and sets CGO_LDFLAGS to the folder holding it. Run it again in a new shell, or put the line it prints in your build environment. On another platform, build the library from a Flezha checkout, as examples/echo shows.

Supported Go versions

Go 1.26 or newer, the release go.mod names. The SDK needs nothing beyond the standard library.

The native library

The core is linked statically through cgo from libflz_core_ffi.a, so a built runner is one executable that loads no core library at run time. That makes cgo, and therefore a C toolchain, a requirement: CGO_ENABLED=0 builds are not supported. The directory holding the library comes from CGO_LDFLAGS. In a Flezha checkout, scripts/build-go-sdk.sh builds it and prints that line:

export CGO_LDFLAGS="-L<flezha>/_build/.target/debug/go"
go test ./...

Outside a checkout, cmd/flz-native (see Install) fetches the library a release publishes and prints the same line.

Writing a relay

type Echo struct{}

func (Echo) Spec() flezha.Spec { return flezha.Spec{Name: "echo"} }

func (Echo) Run(in string, ctx *flezha.Ctx[flezha.NoState, string]) error {
	return ctx.Emit("echo: " + in)
}

type EchoRelay struct{}

func (EchoRelay) Spec() flezha.RelaySpec { return flezha.RelaySpec{FactoryID: "echo"} }

func (EchoRelay) Channels(c *flezha.ChannelBuilder) {
	in := flezha.Input[string](c, "in", flezha.InputOptions{Codec: flezha.Text})
	flezha.Into(in, Echo{}).Output("out")
}

runner, err := flezha.CreateRunner("default")
if err != nil {
	return err
}
if err := runner.RegisterFactory(EchoRelay{}); err != nil {
	return err
}
return runner.Run()

A value leaves a tasklet only through ctx.Emit. A stream wired into a tasklet that cannot take its values fails go build. The guide is docs/internal/sdk/go_sdk.md in the Flezha repository.

Local mode, testing and the relay API

  • Local mode: flezha.StartLocal runs a relay's declared graph in your own process, with no controller. See Local mode.
  • Testing: flezhatest.New, from github.com/flezha/flezha_go_sdk/flezhatest, drives one tasklet and checks its replay. See Check a tasklet on its own.
  • The relay API: a relay's entry channels, answering a caller, flezha.Call to another relay, and runner.Send and runner.Call from application code. See Sending and calling.

Licence

Apache-2.0. See the repository for the full text and third-party notices.

Documentation

Overview

Package flezha is the Flezha Go SDK: a poll host over the shared core, linked in through its C ABI.

The SDK holds no kernel code. Transport, framing, ordering, per-cell exclusivity, durability, retries, timers, routing, the steps driver and reconnects are the core's; this package brings the authoring surface, the codec for Go types, the goroutine host and idiomatic errors. 

type Echo struct{}

func (Echo) Spec() flezha.Spec { return flezha.Spec{Name: "echo"} }

func (Echo) Run(in string, ctx *flezha.Ctx[flezha.NoState, string]) error {
	return ctx.Emit(in)
}

type EchoRelay struct{}

func (EchoRelay) Spec() flezha.RelaySpec { return flezha.RelaySpec{FactoryID: "echo"} }

func (EchoRelay) Channels(c *flezha.ChannelBuilder) {
	flezha.Into(flezha.Input[string](c, "input"), Echo{}).Output("out")
}

runner, err := flezha.CreateRunner("default")
...
err = runner.RegisterFactory(EchoRelay{})
err = runner.Run()
Example (Harness) 
package main

import (
	"fmt"

	flezha "github.com/flezha/flezha_go_sdk"
	"github.com/flezha/flezha_go_sdk/flezhatest"
)

type Order struct {
	Pair string `json:"pair"`
	Qty  int    `json:"qty"`
}

type Book struct {
	Sum int `json:"sum"`
}

type Added struct {
	Qty int `json:"qty"`
}

func (a Added) ApplyTo(b *Book) { b.Sum += a.Qty }

type Totals struct{}

func (Totals) Spec() flezha.Spec { return flezha.Spec{Name: "totals", IdleTTLSeconds: 3600} }

func (Totals) Changes() flezha.Changes[Book] { return flezha.Changes[Book]{"added": Added{}} }

func (Totals) Run(in Order, ctx *flezha.Ctx[Book, int]) error {
	if err := ctx.Change(Added{Qty: in.Qty}); err != nil {
		return err
	}
	return ctx.Emit(ctx.State().Sum)
}

func main() {
	h, err := flezhatest.New[Order, Book, int](Totals{})
	if err != nil {
		panic(err)
	}
	out, err := h.Feed(Order{Pair: "EUR/USD", Qty: 4}, 0, false)
	if err != nil {
		panic(err)
	}
	fmt.Println(out.Outputs, h.State().Sum, h.ReplayEquals())
}

Output:
[4] 4 <nil>
Example (Runner)

The runner a program serves its relay types with; compiled here, run against a controller. 

package main

import (
	flezha "github.com/flezha/flezha_go_sdk"
)

type Order struct {
	Pair string `json:"pair"`
	Qty  int    `json:"qty"`
}

type Book struct {
	Sum int `json:"sum"`
}

type Added struct {
	Qty int `json:"qty"`
}

func (a Added) ApplyTo(b *Book) { b.Sum += a.Qty }

type Totals struct{}

func (Totals) Spec() flezha.Spec { return flezha.Spec{Name: "totals", IdleTTLSeconds: 3600} }

func (Totals) Changes() flezha.Changes[Book] { return flezha.Changes[Book]{"added": Added{}} }

func (Totals) Run(in Order, ctx *flezha.Ctx[Book, int]) error {
	if err := ctx.Change(Added{Qty: in.Qty}); err != nil {
		return err
	}
	return ctx.Emit(ctx.State().Sum)
}

type Orders struct{}

func (Orders) Spec() flezha.RelaySpec { return flezha.RelaySpec{FactoryID: "orders"} }

func (Orders) Channels(c *flezha.ChannelBuilder) {
	orders := flezha.Input[Order](c, "orders").KeyBy(func(o Order) string { return o.Pair })
	flezha.Into(orders, Totals{}).Output("totals")
}

func main() {
	run := func() error {
		runner, err := flezha.CreateRunner("trading")
		if err != nil {
			return err
		}
		if err := runner.RegisterFactory(Orders{}); err != nil {
			return err
		}
		return runner.Run()
	}
	_ = run
}

Index

Examples

Constants

View Source 
const Version = "0.1.1"

Version is this SDK's own released version, stamped into every durable record it writes. Go modules carry no version in go.mod, so this constant is the one source of truth and the release checklist bumps it.

Variables

This section is empty.

Functions

func Effect 

func Effect[T, S, O any](ctx *Ctx[S, O], name string, fn func() (T, error)) (T, error)

Effect runs fn, a call to anything outside the instance, and keeps its result: when this invocation runs again (a crash re-run, a replay, a move) the call returns the kept result and fn does not run. name names the call, counted per invocation. A string or []byte result is kept as it stands and anything else as JSON; an invocation with no input-log offset runs fn each time and keeps nothing.

func NonRetryable 

func NonRetryable(err error) error

NonRetryable marks err as one no retry can change: the item parks at once, with no further attempts.

Types

type CallSpec 

type CallSpec struct {
	Target        string
	Factory       string
	DeadlineMs    int64
	NoCreate      bool
	Channel       string
	AnswerChannel string
	Codec         ChannelCodec
}

CallSpec is a call to another relay at the end of a chain. Target names an instance, or a partition group whose member the key chooses; Factory names a factory whose instance the key names instead, started when absent unless NoCreate. DeadlineMs is required. Channel is the callee's entry channel, AnswerChannel the channel whose output answers (its first output for the request when empty), and Codec the format of the request and the answer, JSON by default.

type Change 

type Change[S any] interface {
	ApplyTo(state *S)
}

Change is one recorded change of a cell's state: ApplyTo folds it in, live and on restore alike, so it must read nothing but the state and itself.

type Changes 

type Changes[S any] map[string]Change[S]

Changes lists a tasklet's change types under explicit names. A record carries the name, never the Go type's, so a type rename keeps every saved record readable. Each value is a zero value of its type.

type ChannelBuilder 

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

ChannelBuilder is what a relay's Channels receives: entry channels come from Input, tasklets are wired with Into, and every step that changes the element type is a package function taking and returning typed streams, so a mismatched edge fails `go build`.

The builder records the chains as written and judges nothing: the core checks the record when the relay registers, and its refusal is the error RegisterFactory returns.

type ChannelCodec 

type ChannelCodec string

ChannelCodec is the payload format a channel names. A channel that names none reads JSON, and an output that names none sends a string or a []byte as it stands and anything else as JSON. 

const (
	JSON  ChannelCodec = "json"
	Text  ChannelCodec = "text"
	Bytes ChannelCodec = "bytes"
)

type Control 

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

Control administers the namespace this runner's connection proved, from ordinary application code: start and stop relays, enable and bind plugins, read what is running. The controller judges every call as the identity the connection established, so the application holds no credential; a user grants a namespace's applications the authority once: 

flz role bind runner@team-orders app team-orders

func (*Control) BindPlugin 

func (c *Control) BindPlugin(name, relayID string, hooks []any, channel string) (any, error)

BindPlugin binds relayID to the enabled plugin name, with optional hooks and channel.

func (*Control) Call 

func (c *Control) Call(verb string, args ...any) (any, error)

Call is any published function, by name, with its arguments other than the namespace, which the controller fills from the connection.

func (*Control) CellState 

func (c *Control) CellState(relayID, cell string) (any, error)

CellState is a relay's live state and the change sequence it was read at. cell is `<tasklet>[/<cell_key>]`; "" reads every cell of the relay.

func (*Control) DisablePlugin 

func (c *Control) DisablePlugin(name string) (any, error)

DisablePlugin disables the plugin name, stopping it wherever it runs.

func (*Control) EnablePlugin 

func (c *Control) EnablePlugin(name, provider string, opts map[string]any, single bool, relayID string, hooks []any) (any, error)

EnablePlugin enables the plugin name of provider (for example relay_websocket) on opts. single runs one copy in the whole cluster; relayID with hooks binds that relay in the same call.

func (*Control) ListPlugins 

func (c *Control) ListPlugins() (any, error)

ListPlugins is the plugins enabled in this namespace, one row each.

func (*Control) ListRelays 

func (c *Control) ListRelays() (any, error)

ListRelays is the relays in this namespace across the cluster: rows, one row each, and unreachable, the controllers that did not answer.

func (*Control) StartRelay 

func (c *Control) StartRelay(factoryID, id string, initValues, opts any) (StartResult, error)

StartRelay starts a relay of factoryID. Without an id the controller generates one, so a start retried after a lost answer starts a second relay; with one, a start of an id already running answers already_running. initValues and opts are the specification `flz relay start` takes, nil for none.

func (*Control) StopRelay 

func (c *Control) StopRelay(relayID string) (any, error)

StopRelay stops a relay, or the whole partition group it names.

func (*Control) UnbindPlugin 

func (c *Control) UnbindPlugin(name, relayID string) (any, error)

UnbindPlugin unbinds relayID from the plugin name, which stays enabled.

type Ctx 

type Ctx[S, O any] struct {
	// contains filtered or unexported fields
}

Ctx is the invocation context: every deterministic effect a tasklet is allowed. It accumulates one item's effects (recorded changes, emitted values, timer operations, the checkpoint) and the runner turns them into the record the core folds; nothing reaches the wire on its own.

A failing call returns its error; Run returning it fails the item, as an uncaught exception does in the other SDKs.

func (*Ctx[S, O]) CancelTimer 

func (c *Ctx[S, O]) CancelTimer(kind TimerKind, at uint64) error

CancelTimer cancels a registered timer by the time its registration answered.

func (*Ctx[S, O]) CellKey 

func (c *Ctx[S, O]) CellKey() (string, bool)

CellKey is the cell's key, and false for a single-cell tasklet.

func (*Ctx[S, O]) Change 

func (c *Ctx[S, O]) Change(change Change[S]) error

Change records one change: journaled, then folded into the state through the same ApplyTo a restore uses. A change type the tasklet's Changes does not list is refused with runtime.change_not_registered.

func (*Ctx[S, O]) Checkpoint 

func (c *Ctx[S, O]) Checkpoint()

Checkpoint closes the open checkpoint interval (checkpoint ack-granularity classes).

func (*Ctx[S, O]) Complete 

func (c *Ctx[S, O]) Complete(result any) error

Complete ends the instance with result once the current iteration commits; a later call replaces an earlier one. The result is sent as an output naming no codec sends a value, and nil is no result.

func (*Ctx[S, O]) Emit 

func (c *Ctx[S, O]) Emit(value O, opts ...EmitOption) error

Emit sends one value on the tasklet's output: to each consumer through its own edge, and to the channel's output or call when it ends in one.

func (*Ctx[S, O]) Now 

func (c *Ctx[S, O]) Now() uint64

Now is the core-evaluated clock for this item, never ambient time.

func (*Ctx[S, O]) Part 

func (c *Ctx[S, O]) Part(value O) error

Part sends value as a part: a preview of the output this invocation is producing, which leaves now rather than when Run returns. It never waits; a part that cannot leave at once is dropped, and a replay sends none.

func (*Ctx[S, O]) Replay 

func (c *Ctx[S, O]) Replay() bool

Replay answers whether this invocation replays an input already processed.

func (*Ctx[S, O]) SpawnSteps 

func (c *Ctx[S, O]) SpawnSteps(factoryID string, payload any, opts ...SpawnOption) (*SpawnHandle, error)

SpawnSteps asks the controller for a steps instance of factoryID with payload as its first input, and answers the handle at once. A replayed invocation sends nothing and its handle answers skipped_on_replay.

func (*Ctx[S, O]) StableID 

func (c *Ctx[S, O]) StableID() string

StableID is a replay-stable identifier, distinct per call within the invocation: the n-th call of a replayed invocation equals the n-th of the original.

func (*Ctx[S, O]) State 

func (c *Ctx[S, O]) State() S

State is a copy of the cell's state: assigning a field of it changes nothing durable. Maps, slices and pointers inside it are shared and not guarded; change state only through Change.

func (*Ctx[S, O]) TimerAfter 

func (c *Ctx[S, O]) TimerAfter(ms uint64) (uint64, error)

TimerAfter registers a processing-time timer ms after Now, and answers its time.

func (*Ctx[S, O]) TimerAtEventTime 

func (c *Ctx[S, O]) TimerAtEventTime(at uint64) (uint64, error)

TimerAtEventTime registers an event-time timer at at, and answers it.

type EmitOption 

type EmitOption func(*emitOptions)

EmitOption qualifies one Emit.

func WithChannel 

func WithChannel(name string) EmitOption

WithChannel emits on the named channel this tasklet produces on.

func WithKey 

func WithKey(key string) EmitOption

WithKey emits under key instead of the edge's own.

func WithMetadata 

func WithMetadata(metadata map[string]string) EmitOption

WithMetadata attaches metadata; a key in the reserved `flz.` namespace is refused.

type Error 

type Error = native.Error

Error is anything Flezha refuses: the controller, the core or this package. Code is what a program branches on (errors.As, then Code), Class the error's class (invalid, not_found, denied, ...), Message one sentence for a reader, Details what the message is about, and Hint the next command to run.

type InputOptions 

type InputOptions struct {
	// The payload format: JSON by default; Text needs a string stream and
	// Bytes a []byte one.
	Codec ChannelCodec
	// What the channel is for, in the relay's API.
	Description string
	// The channel's JSON Schema, as a value json.Marshal writes as an object;
	// nil derives one from T when T is a struct and the codec is JSON.
	Schema any
}

InputOptions are an entry channel's optional settings.

type LocalOutput 

type LocalOutput struct {
	Channel string
	Payload []byte
	// What a deployed run would have put on the wire, `flz.*` stamps included.
	Metadata map[string]string
}

LocalOutput is one value a graph released on a declared output.

type LocalRelay 

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

LocalRelay runs a declared relay graph in this process, with no controller: the deployed machinery (entry decode, edge transforms and keying, one cell per key with its own mailbox, the retry ladder, inter-tasklet delivery) in memory, with nothing durable. Durability, plugins, placement, the dead-letter queue and scale need a controller; the run says so through the logger every time it starts.

func StartLocal 

func StartLocal(relay Relay, opts ...RunnerOption) (*LocalRelay, error)

StartLocal registers relay's graph and starts it locally.

Example 
package main

import (
	"fmt"
	"io"
	"log/slog"
	"time"

	flezha "github.com/flezha/flezha_go_sdk"
)

type Order struct {
	Pair string `json:"pair"`
	Qty  int    `json:"qty"`
}

type Book struct {
	Sum int `json:"sum"`
}

type Added struct {
	Qty int `json:"qty"`
}

func (a Added) ApplyTo(b *Book) { b.Sum += a.Qty }

type Totals struct{}

func (Totals) Spec() flezha.Spec { return flezha.Spec{Name: "totals", IdleTTLSeconds: 3600} }

func (Totals) Changes() flezha.Changes[Book] { return flezha.Changes[Book]{"added": Added{}} }

func (Totals) Run(in Order, ctx *flezha.Ctx[Book, int]) error {
	if err := ctx.Change(Added{Qty: in.Qty}); err != nil {
		return err
	}
	return ctx.Emit(ctx.State().Sum)
}

type Orders struct{}

func (Orders) Spec() flezha.RelaySpec { return flezha.RelaySpec{FactoryID: "orders"} }

func (Orders) Channels(c *flezha.ChannelBuilder) {
	orders := flezha.Input[Order](c, "orders").KeyBy(func(o Order) string { return o.Pair })
	flezha.Into(orders, Totals{}).Output("totals")
}

func main() {
	run, err := flezha.StartLocal(Orders{}, flezha.WithLogger(quietLogger()))
	if err != nil {
		panic(err)
	}
	defer run.Close()
	_ = run.OnOutput("totals", func(out flezha.LocalOutput) { fmt.Println(string(out.Payload)) })
	_ = run.Feed("orders", []byte(`{"pair":"EUR/USD","qty":2}`))
	_ = run.Feed("orders", []byte(`{"pair":"EUR/USD","qty":3}`))
	if err := run.RunUntilIdle(10 * time.Second); err != nil {
		panic(err)
	}
}

func quietLogger() *slog.Logger { return slog.New(slog.NewTextHandler(io.Discard, nil)) }

Output:
2
5

func (*LocalRelay) Close 

func (l *LocalRelay) Close() error

Close ends the run and releases its cells. Idempotent.

func (*LocalRelay) Feed 

func (l *LocalRelay) Feed(channel string, payload []byte) error

Feed feeds one value's wire bytes to a declared entry channel.

func (*LocalRelay) OnOutput 

func (l *LocalRelay) OnOutput(channel string, callback func(LocalOutput)) error

OnOutput delivers every value released on the output channel to callback, in release order. A channel the declaration does not carry is refused.

func (*LocalRelay) RunUntilIdle 

func (l *LocalRelay) RunUntilIdle(budget time.Duration) error

RunUntilIdle drives the graph until it has nothing left to do: take an activation if there is one and answer it, hand on what the outputs released, and when a turn hands nothing back ask the core whether the graph settled. A budget that elapses first is reported, never passed off as done.

type Merge 

type Merge[M any] struct {
	// contains filtered or unexported fields
}

Merge merges several channels into one tasklet, each source lifted into the merged value M by its own mapping.

func From 

func From[T, M any](m *Merge[M], s Stream[T], lift func(T) M) *Merge[M]

From adds one source; its mapping is an ordinary map on the source's chain.

func NewMerge 

func NewMerge[M any](c *ChannelBuilder) *Merge[M]

NewMerge starts a merge on c.

func (*Merge[M]) KeyBy 

func (m *Merge[M]) KeyBy(key func(M) string) *Merge[M]

KeyBy routes every source by a key computed on the merged value.

type NoOutput 

type NoOutput struct{}

NoOutput is the output of a tasklet that emits nothing.

type NoState 

type NoState struct{}

NoState is the state of a stateless tasklet.

type Overflow 

type Overflow string

Overflow is what a channel does when its consumer's mailbox is full. 

const (
	DropNothing Overflow = "drop_nothing"
	DropNewest  Overflow = "drop_newest"
	DropOldest  Overflow = "drop_oldest"
)

type Relay 

type Relay interface {
	Spec() RelaySpec
	Channels(c *ChannelBuilder)
}

Relay is a relay type: its RelaySpec and its channels graph.

func ShortRelay 

func ShortRelay[I, S, O any](factoryID string, fn func(in I, ctx *Ctx[S, O]) (O, bool), opts ...ShortOptions) Relay

ShortRelay declares a one-tasklet relay from a function: it emits the value fn returns when the boolean is true, and nothing when it is false. The tasklet is named factoryID; a stateless one uses NoState.

Example 
package main

import (
	"fmt"
	"io"
	"log/slog"
	"strings"
	"time"

	flezha "github.com/flezha/flezha_go_sdk"
)

func main() {
	upcase := flezha.ShortRelay("upcase", func(in string, _ *flezha.Ctx[flezha.NoState, string]) (string, bool) {
		return strings.ToUpper(in), in != ""
	}, flezha.ShortOptions{Codec: flezha.Text, Output: "out"})
	run, err := flezha.StartLocal(upcase, flezha.WithLogger(quietLogger()))
	if err != nil {
		panic(err)
	}
	defer run.Close()
	_ = run.OnOutput("out", func(out flezha.LocalOutput) { fmt.Println(string(out.Payload)) })
	for _, in := range []string{"", "eur/usd"} {
		_ = run.Feed("input", []byte(in))
	}
	if err := run.RunUntilIdle(10 * time.Second); err != nil {
		panic(err)
	}
}

func quietLogger() *slog.Logger { return slog.New(slog.NewTextHandler(io.Discard, nil)) }

Output:
EUR/USD

func ShortTasklet 

func ShortTasklet[I, S, O any](factoryID string, t Tasklet[I, S, O], opts ...ShortOptions) Relay

ShortTasklet declares a one-tasklet relay from a tasklet type, whose hooks it keeps. The tasklet is named factoryID: its Spec leaves Name empty or writes the same id.

type RelaySpec 

type RelaySpec struct {
	FactoryID     string
	Run           any
	Durable       any
	MaxDurationMs int64
	Description   string
}

RelaySpec is the factory id a relay registers under and its class settings, each written as the settings tree's own data: Run "steps" or map[string]any{"steps": map[string]any{"each": "per_input"}}, Durable false, "local" or map[string]any{"local": map[string]any{"sleep_after": 2000}}. Unwritten, the class runs as a flow and keeps its work locally. A finished instance keeps nothing unless the level sets "keep_finished" to hours, or to true for the controller's window (72 hours by default), so flz relay info can still show its Cells and result. The core reads the tree at registration and refuses a misplaced setting by name.

type Response 

type Response[Req, Ans any] struct {
	Request Req
	Value   Ans
	Err     *Error
	Target  string
	Created bool
}

Response is one outcome of a call, as its Response channel delivers it: the request the call end sent, the answer as Value or the Err the call ended in (a missed deadline is call.timeout), the callee's instance id, and whether this call started the callee.

func (Response[Req, Ans]) Ok 

func (r Response[Req, Ans]) Ok() bool

Ok answers whether the call was answered.

type Routed 

type Routed[D any] struct {
	Key      string
	Document D
}

Routed is one delivery of RouteLatest: the latest document, for one key. It travels as the JSON pair [key, document].

func (Routed[D]) MarshalJSON 

func (r Routed[D]) MarshalJSON() ([]byte, error)

func (*Routed[D]) UnmarshalJSON 

func (r *Routed[D]) UnmarshalJSON(b []byte) error

type Runner 

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

Runner is what an application starts: the relay types it registers and the connection that serves them. One runner per process; N processes are N runners, which the controller's placement already spreads relays across.

func CreateRunner 

func CreateRunner(namespace string, opts ...RunnerOption) (*Runner, error)

CreateRunner creates the runner of namespace. The linked core is checked here: a core of another activation schema is refused, naming both versions, before anything connects.

func (*Runner) AwaitTermination 

func (r *Runner) AwaitTermination() error

AwaitTermination blocks until the runner stops, then releases the core.

func (*Runner) Call 

func (r *Runner) Call(relayID, channel string, value any) (any, error)

Call calls the entry channel of relayID with value, through the controller under the same role check as Send, answering what the relay answers through its outputs: a JSON answer decoded, a text one as a string, a binary one as []byte, and a channel that does not answer as the controller's reply.

func (*Runner) Close 

func (r *Runner) Close()

Close shuts the runner down: every relay's state ships, what is in flight finishes, and AwaitTermination returns.

func (*Runner) Control 

func (r *Runner) Control() *Control

Control administers the namespace this connection proved.

func (*Runner) Namespace 

func (r *Runner) Namespace() string

Namespace is the namespace the runner serves.

func (*Runner) RegisterFactory 

func (r *Runner) RegisterFactory(relay Relay) error

RegisterFactory registers a relay type, a full form or a short form alike. The declaration is judged by the core here, so a malformed one is refused at this call, with the core's code; after Start the controller learns of it at once.

func (*Runner) Relays 

func (r *Runner) Relays() ([]string, error)

Relays is the ids of the relay instances bound on this runner now, read without the controller.

func (*Runner) Run 

func (r *Runner) Run() error

Run is Start followed by AwaitTermination.

func (*Runner) Send 

func (r *Runner) Send(relayID, channel string, message []byte, opts ...SendOption) (*SendVerdict, error)

Send sends message to the entry channel of relayID, through the controller under the same role check as Control, answering its verdict; a Local send answers nil.

func (*Runner) Start 

func (r *Runner) Start() error

Start connects and starts serving: everything up to the first poll, and nothing that blocks after it. Go owns SIGTERM and SIGINT and hands the first to the core, which drains and exits; a closed standard input ends the runner the same way.

func (*Runner) UnregisterFactory 

func (r *Runner) UnregisterFactory(relay Relay) error

UnregisterFactory undoes a registration. After Start the core refuses a relay type with live instances on this runner.

type RunnerOption 

type RunnerOption func(*runnerOptions)

RunnerOption configures CreateRunner and StartLocal.

func WithLogger 

func WithLogger(logger *slog.Logger) RunnerOption

WithLogger hands the runner the application's logger: the SDK's diagnostics and the core's notices go there, and nothing reaches standard error past it.

type SendOption 

type SendOption func(*sendOptions)

SendOption qualifies one Send.

func Local 

func Local() SendOption

Local sends in process, without the controller: the message is on no disk, is lost if the runner stops before handling it, and the relay's input log does not cover it.

func MsgID 

func MsgID(id string) SendOption

MsgID stamps the message as flz.msg_id, so a relay whose input log still holds that id absorbs the repeat.

func SendKey 

func SendKey(key string) SendOption

SendKey sends to the partition group member whose range owns key.

type SendVerdict 

type SendVerdict struct {
	To      string `json:"to"`
	Verdict string `json:"verdict"`
}

SendVerdict is what the controller did with a message: delivered, absorbed or suspended, on the relay it reached.

type ShortOptions 

type ShortOptions struct {
	Input         string
	Codec         ChannelCodec
	Output        string
	Run           any
	Durable       any
	MaxDurationMs int64
	Description   string
}

ShortOptions are a short form's channels and class settings: the entry channel ("input" when empty) and its codec, the plugin output the tasklet's output reaches (none when empty), and the RelaySpec settings.

type SpawnHandle 

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

SpawnHandle is the answer to one SpawnSteps request. Read it inside the invocation that made it: once that invocation completes the core releases the handle, a failure nobody read is logged once instead, and a later read fails. Reading is optional.

func (*SpawnHandle) TryResult 

func (h *SpawnHandle) TryResult() (*SpawnResult, error)

TryResult is the answer if it has arrived, or nil while it is pending. It never waits.

func (*SpawnHandle) Wait 

func (h *SpawnHandle) Wait() (SpawnResult, error)

Wait waits for the controller's answer, up to the core's own bound. A spawn that reached no instance fails with the controller's code (such as relay.already_running) or the core's (spawn.timeout, spawn.unavailable).

type SpawnOption 

type SpawnOption func(*spawnOptions)

SpawnOption qualifies one SpawnSteps.

func IfRunning 

func IfRunning(policy string) SpawnOption

IfRunning says what a running instance under SpawnID gets: "deliver" (the default) feeds it the payload, "error" leaves it alone and the read fails.

func SpawnID 

func SpawnID(id string) SpawnOption

SpawnID names the instance, so a reissued request reaches the same one.

type SpawnOutcome 

type SpawnOutcome string

SpawnOutcome is what a spawn request did. 

const (
	Spawned         SpawnOutcome = "spawned"
	Delivered       SpawnOutcome = "delivered"
	SkippedOnReplay SpawnOutcome = "skipped_on_replay"
)

type SpawnResult 

type SpawnResult struct {
	ID      string       `json:"id"`
	Outcome SpawnOutcome `json:"outcome"`
}

SpawnResult is the instance a spawn request reached, and how.

type Spec 

type Spec struct {
	Name string
	// A static key set; the tasklet implements CellsFor to route an input.
	CellKeys []string
	// Evict an idle dynamic cell after this long; 0 never.
	IdleTTLSeconds int64
	MaxCells       int
	// Serve the cell's change sequence as Now().
	LogicalClock bool
	// The per-cell mailbox capacity; nil is the core's default. A pointer, so
	// a capacity of 0 can be written and refused by name.
	MailboxCapacity *int
	// The schema version the state and change records are written at; 0 is 1.
	SchemaVersion uint32
	// The oldest version still read; 0 is SchemaVersion.
	MinReadableSchemaVersion uint32
	// The pending timers one cell may hold; 0 is the core's default.
	MaxTimersPerCell int
	// The cell log size that asks for a snapshot; 0 is the controller's default.
	CompactAfterBytes int64
}

Spec is everything a tasklet declares that is not behaviour, read once when the relay wiring it registers, never per message. Name is required: it is part of every cell's durable path, so renaming the Go type never moves saved state. A zero field leaves the core's default.

The cell mode is not declared here: a tasklet fed through KeyBy is dynamic, one with CellKeys static, and any other single.

type StartResult 

type StartResult struct {
	ID      string `json:"id"`
	Outcome string `json:"outcome"`
}

StartResult is the relay a start reached, and whether it started it.

type Stream 

type Stream[T any] struct {
	// contains filtered or unexported fields
}

Stream is one edge under construction, carrying values of type T. It is a value: every call answers a new stream, so a shared prefix can branch without its arms colliding.

func Call 

func Call[Req, Ans any](s Stream[Req], spec CallSpec) Stream[Response[Req, Ans]]

Call ends this edge's chain in a call to another relay, and answers the stream of its Response channel, the entry channel `<channel>.response`. Every value reaching the call, after this chain's transforms and key, is one request. A call is not an edge, so the Response stream may feed the calling tasklet.

func FlatMap 

func FlatMap[T, U any](s Stream[T], f func(T) []U) Stream[U]

FlatMap transforms one value into many; each reaches the consumer on its own.

func Input 

func Input[T any](c *ChannelBuilder, name string, opts ...InputOptions) Stream[T]

Input declares a controller-fed entry channel whose payloads decode into T.

func InputDecoded 

func InputDecoded[T any](c *ChannelBuilder, name string, decode func([]byte) (T, error), opts ...InputOptions) Stream[T]

InputDecoded declares an entry channel whose payloads decode is the author's own.

func Into 

func Into[I, S, O any](s Stream[I], t Tasklet[I, S, O]) Stream[O]

Into wires the stream into a tasklet whose input is the stream's element type, and answers the tasklet's output stream.

func IntoNamed 

func IntoNamed[I, S, O any](s Stream[I], t Tasklet[I, S, O], nodeName string) Stream[O]

IntoNamed is Into under an explicit node identity, for a second instance of one tasklet type.

func Map 

func Map[T, U any](s Stream[T], f func(T) U) Stream[U]

Map transforms every value on the edge, producer-side.

func MergeInto 

func MergeInto[S, O, M any](m *Merge[M], t Tasklet[M, S, O]) Stream[O]

MergeInto closes every source on one tasklet node, and answers its output.

func MergeIntoNamed 

func MergeIntoNamed[S, O, M any](m *Merge[M], t Tasklet[M, S, O], nodeName string) Stream[O]

MergeIntoNamed is MergeInto under an explicit node identity.

func RouteLatest 

func RouteLatest[D, T any](c *ChannelBuilder, documents Stream[D], traffic Stream[T], key func(T) string) Stream[Routed[D]]

RouteLatest turns a relay-wide document stream into per-key deliveries: every key seen on traffic receives each distinct document once, on that key's own next traffic item. Delivering on the key's own traffic keeps every key one this member owns after a rescale; the price is one tick of latency, and a key with no traffic receives nothing.

func RouteLatestNamed 

func RouteLatestNamed[D, T any](c *ChannelBuilder, documents Stream[D], traffic Stream[T], key func(T) string, nodeName string) Stream[Routed[D]]

RouteLatestNamed is RouteLatest under an explicit node identity, for a second router in one relay.

func (Stream[T]) Branch 

func (s Stream[T]) Branch(arms ...func(Stream[T])) Stream[T]

Branch fans out: each arm receives this same edge, so Branch builds exactly the topology the separate statements would.

func (Stream[T]) Filter 

func (s Stream[T]) Filter(pred func(T) bool) Stream[T]

Filter drops the values pred rejects; a dropped value is consumed and reaches no handler.

func (Stream[T]) KeyBy 

func (s Stream[T]) KeyBy(key func(T) string) Stream[T]

KeyBy routes by a computed key: the consumer becomes a dynamic keyed tasklet.

func (Stream[T]) Keyed 

func (s Stream[T]) Keyed() Stream[T]

Keyed stamps `flz.cell` / `flz.seq` / `flz.key` onto this channel's plugin frames.

func (Stream[T]) Named 

func (s Stream[T]) Named(name string) Stream[T]

Named overrides the channel name this edge creates.

func (Stream[T]) OnFull 

func (s Stream[T]) OnFull(policy Overflow) Stream[T]

OnFull sets what the channel does when its consumer's mailbox is full.

func (Stream[T]) Output 

func (s Stream[T]) Output(name string, codec ...ChannelCodec) Stream[T]

Output terminates this edge's channel on the plugin layer. A name written here is the channel's name; "" leaves it derived. Without a codec a string or []byte is sent as it stands and anything else as JSON.

func (Stream[T]) OutputEncoded 

func (s Stream[T]) OutputEncoded(name string, encode func(T) ([]byte, error)) Stream[T]

OutputEncoded is Output with the plugin frame's bytes written by encode.

func (Stream[T]) RetryBackoffMs 

func (s Stream[T]) RetryBackoffMs(ms int) Stream[T]

RetryBackoffMs is the channel's backoff between attempts.

func (Stream[T]) RetryMax 

func (s Stream[T]) RetryMax(attempts int) Stream[T]

RetryMax is the channel's attempt ceiling.

type Tasklet 

type Tasklet[I, S, O any] interface {
	Spec() Spec
	Run(in I, ctx *Ctx[S, O]) error
}

Tasklet is one processing step: a type with a Spec and a Run. A value leaves it only through ctx.Emit. Run returning nil completes the item; any other error is retried up the channel's retry ladder, and one wrapped by NonRetryable parks at once. A panic is recovered and retried as an error carrying its text.

A hook is declared by implementing it: InitState, Changes, OnTimer, EventTime, CellsFor, Compensate, MigrateState and MigrateChange, each with the tasklet's own I, S and O.

type TimerFiring 

type TimerFiring struct {
	Kind        TimerKind
	ScheduledAt uint64
}

TimerFiring is a fired timer, as OnTimer receives it. ctx.Now() inside the hook equals ScheduledAt, so a chained TimerAfter is replay-stable.

type TimerKind 

type TimerKind string

TimerKind is the clock a timer was scheduled against. 

const (
	EventTime      TimerKind = "event_time"
	ProcessingTime TimerKind = "processing_time"
)

Source Files

View all Source files

Directories

Path Synopsis
cmd
flz-native command
Command flz-native downloads the prebuilt core library the SDK links through cgo, checks it against the release's checksums, and prints the CGO_LDFLAGS line that points Go at it:
 Command flz-native downloads the prebuilt core library the SDK links through cgo, checks it against the release's checksums, and prints the CGO_LDFLAGS line that points Go at it:
Package flezhatest drives one tasklet without a controller, for an author's own tests: the SDK's own dispatch path over a one-tasklet relay, so what a test proves is what the runner does.
 Package flezhatest drives one tasklet without a controller, for an author's own tests: the SDK's own dispatch path over a one-tasklet relay, so what a test proves is what the runner does.
internal
check
Package check is the seam the conformance run judges a relay through: the SDK fills Derived at init, and the test app reads the declaration the core derived for a relay, with nothing registered and no runner.
 Package check is the seam the conformance run judges a relay through: the SDK fills Derived at init, and the test app reads the declaration the core derived for a relay, with nothing registered and no runner.
codec
Package codec is the `Activation` / `ActivationResult` boundary contract in Go: postcard, field for field as `crates/flz_core/src/schema.rs` declares it, at native.SchemaVersion.
 Package codec is the `Activation` / `ActivationResult` boundary contract in Go: postcard, field for field as `crates/flz_core/src/schema.rs` declares it, at native.SchemaVersion.
contract
Package contract finds the contract set: the fixtures under `test/fixtures` and the error catalogue every SDK is held to, which live in the core's tree.
 Package contract finds the contract set: the fixtures under `test/fixtures` and the error catalogue every SDK is held to, which live in the core's tree.
harness
Package harness is the seam between the SDK and its test harness: the SDK fills New at init, and package flezhatest drives one tasklet through it with the SDK's own dispatch path, never a second implementation of it.
 Package harness is the seam between the SDK and its test harness: the SDK fills New at init, and package flezhatest drives one tasklet through it with the SDK's own dispatch path, never a second implementation of it.
native
Package native is the Go binding of the core's C ABI: one function per entry point of `test/fixtures/abi_surface.txt`, called through cgo against the static `libflz_core_ffi.a`.
 Package native is the Go binding of the core's C ABI: one function per entry point of `test/fixtures/abi_surface.txt`, called through cgo against the static `libflz_core_ffi.a`.
The conformance run: every row of `test/fixtures/reference_topologies.txt` driven through THIS SDK's builder, so what is measured is what an author of a Go relay writes.
 The conformance run: every row of `test/fixtures/reference_topologies.txt` driven through THIS SDK's builder, so what is measured is what an author of a Go relay writes.

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.