cluster

Documentation

Index

• func NewSchedulerEngine(engine *workflow.Engine) scheduler.Engine
• type ClusterExecutor
  • func NewClusterExecutor(engine scheduler.Engine, selfID string, registry contract.NodeRegistry, ...) *ClusterExecutor
  • func (c *ClusterExecutor) Execute(ctx context.Context, task spi.ScheduledTask, target string)
• type Config
• type SchedulerRPCClient
  • func NewSchedulerRPCClient(auth dispatch.PeerAuth, timeout time.Duration) *SchedulerRPCClient
  • func (c *SchedulerRPCClient) AllowLoopbackForTesting() *SchedulerRPCClient
  • func (c *SchedulerRPCClient) ExecuteScheduledTask(ctx context.Context, addr string, task spi.ScheduledTask) error
• type SchedulerRPCHandler
  • func NewSchedulerRPCHandler(engine scheduler.Engine, auth dispatch.PeerAuth) *SchedulerRPCHandler
  • func (h *SchedulerRPCHandler) Register(mux *http.ServeMux)
• type SchedulerTaskRequest
• type SchedulerTaskResponse

Functions

func NewSchedulerEngine

func NewSchedulerEngine(engine *workflow.Engine) scheduler.Engine

NewSchedulerEngine adapts a real *workflow.Engine to scheduler.Engine for use with LocalExecutor, ClusterExecutor, and SchedulerRPCHandler.

Types

type ClusterExecutor

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

ClusterExecutor implements scheduler.Executor: it fires a ScheduledTask locally when the coordinator picked this node as the dispatch target, or forwards it to the target peer over the existing PeerAuth-authenticated dispatch channel — the same AEAD-wrapped transport callout dispatch uses (internal/cluster/dispatch/forwarder.go) — when the target is a peer. There is no unauthenticated forwarding path.

A peer-forward failure (unresolvable target, unreachable peer, auth rejected) is logged and dropped — it never falls back to firing locally, which would silently defeat the distribution strategy the coordinator chose and could concentrate load back onto the coordinator node during a partition. The scheduler's at-least-once redispatch (RedispatchBackoff, internal/scheduler/service.go) covers it: the row is still due on a later scan and may then land on a different, live target.

func NewClusterExecutor

func NewClusterExecutor(engine scheduler.Engine, selfID string, registry contract.NodeRegistry, client *SchedulerRPCClient) *ClusterExecutor

NewClusterExecutor constructs a ClusterExecutor. registry and client may be nil for a single-node/cluster-disabled deployment where every dispatch target is always selfID (see scheduler.Self distribution) — Execute never reaches the forward branch in that configuration, but nil is handled defensively regardless.

func (*ClusterExecutor) Execute

func (c *ClusterExecutor) Execute(ctx context.Context, task spi.ScheduledTask, target string)

Execute implements scheduler.Executor.

type Config

type Config struct {
	Enabled  bool
	NodeID   string
	NodeAddr string
	// GRPCNodeAddr is this node's advertised gRPC endpoint (host:port, no scheme).
	// When set it is gossiped to peers; when empty peers derive it from NodeAddr's
	// host plus their own configured gRPC port.
	GRPCNodeAddr           string
	GossipAddr             string
	SeedNodes              []string
	StabilityWindow        time.Duration
	TxTTL                  time.Duration
	TxReapInterval         time.Duration
	ProxyTimeout           time.Duration
	OutcomeTTL             time.Duration
	HMACSecret             []byte
	DispatchWaitTimeout    time.Duration
	DispatchForwardTimeout time.Duration
	TxTokenTTL             time.Duration
	// DispatchAllowLoopback opts the inter-node dispatch HTTP forwarder out of
	// its loopback-address SSRF guard so multi-node tests can run every node on
	// 127.0.0.1 and still forward processor/criteria dispatch between them.
	// Sourced from CYODA_DISPATCH_ALLOW_LOOPBACK_FOR_TESTING; defaults false.
	// Never enable in production — it re-opens the SSRF pivot the guard closes.
	DispatchAllowLoopback bool
}

type SchedulerRPCClient

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

SchedulerRPCClient forwards ExecuteScheduledTask calls to a peer over the PeerAuth-authenticated channel. Mirrors dispatch.HTTPForwarder.ForwardCallout's sign/POST/decode shape (internal/cluster/dispatch/forwarder.go) so it reuses the exact same auth implementation instance the app wires for callout dispatch; it is a separate ~small type rather than a method on HTTPForwarder because its request/response payload is scheduler-specific and DispatchForwarder's interface is scoped to callout dispatch.

func NewSchedulerRPCClient

func NewSchedulerRPCClient(auth dispatch.PeerAuth, timeout time.Duration) *SchedulerRPCClient

NewSchedulerRPCClient constructs a SchedulerRPCClient. auth must be the same PeerAuth instance (shared secret) the peer's SchedulerRPCHandler verifies against — in production, the identical instance passed to dispatch.NewDispatchHandler for processor/criteria dispatch.

func (*SchedulerRPCClient) AllowLoopbackForTesting

func (c *SchedulerRPCClient) AllowLoopbackForTesting() *SchedulerRPCClient

AllowLoopbackForTesting opts the client out of the loopback SSRF guard — see dispatch.HTTPForwarder.AllowLoopbackForTesting. Never call this in production. Returns the receiver for fluent construction.

func (*SchedulerRPCClient) ExecuteScheduledTask

func (c *SchedulerRPCClient) ExecuteScheduledTask(ctx context.Context, addr string, task spi.ScheduledTask) error

ExecuteScheduledTask POSTs task to the peer at addr's scheduled-task route, authenticated via the wrapped PeerAuth. The call is fire-and-forget from the coordinator's point of view — a non-nil error means the peer could not be reached or rejected the request; the caller (ClusterExecutor) logs and drops it rather than retrying inline, relying on the scan loop's at-least-once redispatch.

type SchedulerRPCHandler

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

SchedulerRPCHandler serves the peer-authenticated ExecuteScheduledTask route. Mirrors dispatch.DispatchHandler's auth pattern exactly — the same PeerAuth.Verify-or-403 gate, the same "never log the task payload beyond ids" discipline — so the scheduled-task peer surface carries the identical security posture as processor/criteria dispatch (Gate 3: no new unauthenticated cluster surface).

func NewSchedulerRPCHandler

func NewSchedulerRPCHandler(engine scheduler.Engine, auth dispatch.PeerAuth) *SchedulerRPCHandler

NewSchedulerRPCHandler constructs a SchedulerRPCHandler. auth must be the same PeerAuth instance peers sign with via SchedulerRPCClient.

func (*SchedulerRPCHandler) Register

func (h *SchedulerRPCHandler) Register(mux *http.ServeMux)

Register registers the scheduled-task route on mux.

type SchedulerTaskRequest

type SchedulerTaskRequest struct {
	Task spi.ScheduledTask `json:"task"`
}

SchedulerTaskRequest is the cross-node payload for ExecuteScheduledTask.

type SchedulerTaskResponse

type SchedulerTaskResponse struct {
	Success bool   `json:"success"`
	Error   string `json:"error,omitempty"`
}

SchedulerTaskResponse acks a peer-delegated fire. The coordinator does not depend on its content — Execute is fire-and-forget (design doc §6.2) — but Success/Error are populated for diagnostics; Error is sanitized, never the raw underlying error text.