synapsys

README

Synapsys Go SDK

Framework-neutral Go integration for registering business processes with Synapsys Core and applying remote lifecycle control.

The SDK does not schedule functions. Core decides when a process starts or stops:

• Endless processes are listeners, consumers, or loops that remain active.
• Progressive processes perform finite work once and return naturally.

Everything uses one outbound heartbeat request. The worker never opens a port.

Requirements

• Go 1.22 or later
• A reachable Synapsys Core instance
• A worker token created in Core

Install

go get github.com/basamento/synapsys-go-sdk@v0.1.0

Configure and start

Set the two deployment-specific values:

export SYNAPSYS_CORE_URL=https://core.example.com
export SYNAPSYS_CORE_TOKEN=syn_your_worker_token

Then register processes and start one explicit worker:

package main

import (
	"context"
	"log"
	"time"

	"github.com/basamento/synapsys-go-sdk"
)

func main() {
	worker, err := synapsys.New(
		synapsys.WithWorkerName("billing-api"),
	)
	if err != nil {
		log.Fatal(err)
	}

	err = worker.Register(
		synapsys.EndlessContext("invoice-poller", func(ctx context.Context) error {
			for {
				if err := pollInvoices(ctx); err != nil {
					return err
				}
				select {
				case <-ctx.Done():
					return ctx.Err()
				case <-time.After(5 * time.Second):
				}
			}
		}),
		synapsys.Progressive("monthly-report", buildMonthlyReport),
	)
	if err != nil {
		log.Fatal(err)
	}
	if err := worker.Start(); err != nil {
		log.Fatal(err)
	}

	// Integrate this with the application's normal shutdown path.
	defer func() {
		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer cancel()
		_ = worker.Stop(ctx)
	}()
	select {}
}

Start succeeds when Core is unavailable unless fail-fast is enabled. The heartbeat continues retrying and never stops running business work.

Process declarations

Every body returns an error. A non-nil error or a recovered panic produces the failed state. A body can be an ordinary function or method value from a package that has no Synapsys dependency.

Endless with cooperative cancellation

synapsys.EndlessContext("queue-consumer", func(ctx context.Context) error {
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case message := <-messages:
			consume(message)
		}
	}
})

Endless with an existing stop mechanism

synapsys.Endless(
	"legacy-listener",
	listener.Serve,
	synapsys.WithOnStop(listener.Close),
)

The hook runs when Stop is accepted and can close a socket or listener to unblock the body.

Progressive finite work

synapsys.Progressive("monthly-report", reports.Build)

Progressive with in-flight cancellation

synapsys.ProgressiveContext("bulk-import", func(ctx context.Context) error {
	for _, row := range rows {
		if err := ctx.Err(); err != nil {
			return err
		}
		if err := importRow(ctx, row); err != nil {
			return err
		}
	}
	return nil
})

Cancellation is cooperative. Go cannot terminate a goroutine. If a body ignores its context and has no external stop mechanism, it remains stopping until it returns. A panic in a goroutine created by the business function is also outside the SDK's recovery boundary and can terminate the application.

Process logs

Go has no supported goroutine-local mechanism that can attribute global os.Stdout and os.Stderr writes to concurrent processes. The SDK therefore does not replace those streams. Use a process-scoped logger or writer.

Structured logging with slog:

processLog := worker.Logger("invoice-poller")

synapsys.EndlessContext("invoice-poller", func(ctx context.Context) error {
	processLog.Info("poll complete", "invoices", 4)
	return runPoller(ctx)
})

The logger passes every record to the application's configured slog.Logger and also sends Info-or-higher records produced during an active execution to Core.

For the standard log package or third-party loggers accepting io.Writer:

processOutput := worker.Writer("invoice-poller", synapsys.Stdout, os.Stdout)
processLog := log.New(processOutput, "", log.LstdFlags)

The writer is a tee: application output remains intact. Output outside an active execution is written normally but is not sent to Core. Console capture transmits the resulting process output to Core; disable it if those messages may contain sensitive data.

Configuration

Options override environment variables. Durations in environment variables must carry units accepted by time.ParseDuration; a unitless string is rejected.

Environment variable	Option	Default
SYNAPSYS_CORE_URL	WithCoreURL	required
SYNAPSYS_CORE_TOKEN	WithCoreToken	none; Core normally rejects it
SYNAPSYS_WORKER_NAME	WithWorkerName	executable name
SYNAPSYS_ENABLED	WithEnabled	true
SYNAPSYS_HOST	WithHost	machine hostname
SYNAPSYS_HEARTBEAT_INTERVAL	WithHeartbeatInterval	5s
SYNAPSYS_FAIL_FAST	WithFailFast	false
SYNAPSYS_CONNECT_TIMEOUT	WithConnectTimeout	2s
SYNAPSYS_REQUEST_TIMEOUT	WithRequestTimeout	5s
SYNAPSYS_CAPTURE_CONSOLE	WithConsoleCapture	true
—	WithLogger	slog.Default()

enabled=false is a complete runtime no-op: no validation, network calls, heartbeat goroutine, or capture. Process declaration errors are still reported.

The SDK warns when a token is sent over non-loopback plaintext HTTP. Redirects are never followed, so a bearer token cannot be forwarded to another host.

Shutdown

Use a bounded context from the application's shutdown path:

ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()

if err := worker.Stop(ctx); err != nil {
	log.Printf("Synapsys processes exceeded shutdown deadline: %v", err)
}

The SDK stops normal heartbeats, requests cooperative process stops, waits up to the deadline, and makes one best-effort final heartbeat carrying final states and cleanup logs. Commands in the final response are ignored because shutdown must not start new work.

Development

gofmt -w .
go vet ./...
go test ./...
go test -race ./...

The module intentionally has no runtime dependencies.

API stability

Releases follow semantic versioning. While the module is v0.x, public API changes may occur in a minor release and are called out in release notes. After v1.0.0, incompatible public API or wire-behavior changes require a new major module path; compatible additions and fixes use minor and patch releases.

License

Apache License 2.0. See LICENSE.

Documentation

Overview

Package synapsys connects ordinary Go business functions to Synapsys Core.

The package is framework-neutral. A Worker reports registered processes to Core, applies token-gated start and stop commands, and optionally forwards process-scoped logs. Synapsys does not schedule functions: an Endless process keeps itself alive until it observes cancellation, while a Progressive process runs once and returns naturally.

Index

• type ConfigError
  • func (e *ConfigError) Error() string
• type LogSource
• type Option
  • func WithConnectTimeout(value time.Duration) Option
  • func WithConsoleCapture(value bool) Option
  • func WithCoreToken(value string) Option
  • func WithCoreURL(value string) Option
  • func WithEnabled(enabled bool) Option
  • func WithFailFast(value bool) Option
  • func WithHeartbeatInterval(value time.Duration) Option
  • func WithHost(value string) Option
  • func WithLogger(value *slog.Logger) Option
  • func WithRequestTimeout(value time.Duration) Option
  • func WithWorkerName(value string) Option
• type Process
  • func Endless(name string, run func() error, options ...ProcessOption) Process
  • func EndlessContext(name string, run func(context.Context) error, options ...ProcessOption) Process
  • func Progressive(name string, run func() error, options ...ProcessOption) Process
  • func ProgressiveContext(name string, run func(context.Context) error, options ...ProcessOption) Process
• type ProcessOption
  • func WithOnStop(hook func() error) ProcessOption
• type ProcessState
• type ProcessStatus
• type ProcessType
• type Settings
• type StartupError
  • func (e *StartupError) Error() string
  • func (e *StartupError) Unwrap() error
• type Worker
  • func New(options ...Option) (*Worker, error)
  • func (w *Worker) Logger(processName string) *slog.Logger
  • func (w *Worker) Processes() []ProcessStatus
  • func (w *Worker) Register(declarations ...Process) error
  • func (w *Worker) Settings() Settings
  • func (w *Worker) Start() error
  • func (w *Worker) Stop(ctx context.Context) error
  • func (w *Worker) Writer(processName string, source LogSource, next io.Writer) io.Writer

Examples

• Worker.Register
• Worker.Stop

Types

type ConfigError

type ConfigError struct{ Message string }

ConfigError reports invalid SDK configuration or process declarations.

func (*ConfigError) Error

func (e *ConfigError) Error() string

type LogSource

type LogSource string

LogSource identifies the conventional stream represented by a captured line.

const (
	Stdout LogSource = "stdout"
	Stderr LogSource = "stderr"
)

type Option

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

Option configures a Worker. Options override environment variables.

func WithConnectTimeout

func WithConnectTimeout(value time.Duration) Option

WithConnectTimeout bounds TCP connection establishment.

func WithConsoleCapture

func WithConsoleCapture(value bool) Option

WithConsoleCapture enables or disables explicit process Logger and Writer capture. Go does not replace the process-global os.Stdout or os.Stderr streams.

func WithCoreToken

func WithCoreToken(value string) Option

WithCoreToken sets the bearer token. Environment-based secrets are preferred.

func WithCoreURL

func WithCoreURL(value string) Option

WithCoreURL sets the Synapsys Core base URL.

func WithEnabled

func WithEnabled(enabled bool) Option

WithEnabled completely disables or enables Synapsys integration.

func WithFailFast

func WithFailFast(value bool) Option

WithFailFast makes an unreachable Core fail Worker.Start.

func WithHeartbeatInterval

func WithHeartbeatInterval(value time.Duration) Option

WithHeartbeatInterval sets the positive whole-second heartbeat interval.

func WithHost

func WithHost(value string) Option

WithHost sets the host reported to Core.

func WithLogger

func WithLogger(value *slog.Logger) Option

WithLogger supplies the application's slog logger for SDK diagnostics and process-scoped logger passthrough.

func WithRequestTimeout

func WithRequestTimeout(value time.Duration) Option

WithRequestTimeout bounds a complete Core request.

func WithWorkerName

func WithWorkerName(value string) Option

WithWorkerName sets the stable identity reported across restarts.

type Process

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

Process is a declaration registered with a Worker. Its fields are intentionally private so invalid combinations cannot be built without using Endless, EndlessContext, Progressive, or ProgressiveContext.

func Endless

func Endless(name string, run func() error, options ...ProcessOption) Process

Endless declares long-lived logic with its own stop mechanism.

func EndlessContext

func EndlessContext(name string, run func(context.Context) error, options ...ProcessOption) Process

EndlessContext declares long-lived logic that observes cooperative cancellation.

func Progressive

func Progressive(name string, run func() error, options ...ProcessOption) Process

Progressive declares finite work that does not require in-flight cancellation.

func ProgressiveContext

func ProgressiveContext(name string, run func(context.Context) error, options ...ProcessOption) Process

ProgressiveContext declares finite work that observes cooperative cancellation.

type ProcessOption

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

ProcessOption configures one process declaration.

func WithOnStop

func WithOnStop(hook func() error) ProcessOption

WithOnStop installs an optional cleanup hook. Endless hooks run as soon as a stop is accepted so they can unblock legacy listeners. Progressive hooks run after the body returns from a requested stop.

type ProcessState

type ProcessState string

ProcessState is the state reported to Synapsys Core.

const (
	StateIdle     ProcessState = "idle"
	StateRunning  ProcessState = "running"
	StateStopping ProcessState = "stopping"
	StateFailed   ProcessState = "failed"
)

type ProcessStatus

type ProcessStatus struct {
	Name     string
	Type     ProcessType
	State    ProcessState
	AckToken int64
	RunID    *int64
}

ProcessStatus is a concurrency-safe snapshot of one registered process.

type ProcessType

type ProcessType string

ProcessType identifies a managed process's execution model.

const (
	// TypeEndless identifies business logic that remains running until stopped.
	TypeEndless ProcessType = "endless"
	// TypeProgressive identifies finite business logic that returns naturally.
	TypeProgressive ProcessType = "progressive"
)

type Settings

type Settings struct {
	Enabled           bool
	CoreURL           string
	WorkerName        string
	Host              string
	HeartbeatInterval time.Duration
	FailFast          bool
	ConnectTimeout    time.Duration
	RequestTimeout    time.Duration
	CaptureConsole    bool
}

Settings is the resolved, non-secret worker configuration.

type StartupError

type StartupError struct{ Cause error }

StartupError reports a fail-fast startup failure.

func (*StartupError) Error

func (e *StartupError) Error() string

func (*StartupError) Unwrap

func (e *StartupError) Unwrap() error

type Worker

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

Worker owns an explicit set of Synapsys processes and their lifecycle. A Worker is safe for concurrent observation and logging.

func New

func New(options ...Option) (*Worker, error)

New constructs a framework-neutral worker. Environment variables are resolved before options, so explicit options win. Network access begins only at Start.

func (*Worker) Logger

func (w *Worker) Logger(processName string) *slog.Logger

Logger returns a process-scoped structured logger. Records continue to the application's configured logger and, while that process has an active run, are also queued for Core.

func (*Worker) Processes

func (w *Worker) Processes() []ProcessStatus

Processes returns process snapshots in registration order.

func (*Worker) Register

func (w *Worker) Register(declarations ...Process) error

Register adds process declarations atomically. Every process must be registered before Start, and names must be unique within the worker.

Example

package main

import (
	"context"
	"fmt"

	"github.com/basamento/synapsys-go-sdk"
)

func main() {
	worker, err := synapsys.New(synapsys.WithEnabled(false))
	if err != nil {
		panic(err)
	}
	err = worker.Register(
		synapsys.EndlessContext("listener", func(ctx context.Context) error {
			<-ctx.Done()
			return ctx.Err()
		}),
		synapsys.Progressive("report", func() error { return nil }),
	)
	if err != nil {
		panic(err)
	}

	for _, process := range worker.Processes() {
		fmt.Println(process.Name, process.Type, process.State)
	}
}

Output:
listener endless idle
report progressive idle

func (*Worker) Settings

func (w *Worker) Settings() Settings

Settings returns resolved configuration with the bearer token omitted.

func (*Worker) Start

func (w *Worker) Start() error

Start validates configuration, performs one bounded health probe, and starts the heartbeat loop. Core unavailability is non-fatal unless fail-fast is set.

func (*Worker) Stop

func (w *Worker) Stop(ctx context.Context) error

Stop cooperatively stops processes, waits up to ctx's deadline, sends one best-effort final state/log heartbeat, and releases transport resources. It is idempotent; final-heartbeat failures are logged and never returned.

Example

package main

import (
	"context"
	"time"

	"github.com/basamento/synapsys-go-sdk"
)

func main() {
	worker, err := synapsys.New(synapsys.WithEnabled(false))
	if err != nil {
		panic(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	_ = worker.Stop(ctx)
}

func (*Worker) Writer

func (w *Worker) Writer(processName string, source LogSource, next io.Writer) io.Writer

Writer returns a process-scoped tee. Bytes always go to next; complete lines written during an active run are also queued for Core. The caller owns next.