orchestration

package
v0.1.0-beta.2 Latest Latest
Warning

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

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

Documentation

Index

Constants

This section is empty.

Variables

View Source
var (
	// ErrMissingLeaseID marks an orchestration message that requires a
	// lease identifier but carries none.
	ErrMissingLeaseID = errors.New("orchestration message missing lease ID")

	// ErrInvalidPayload marks a request whose JSON payload does not
	// match the schema for its message type.
	ErrInvalidPayload = errors.New("invalid orchestration message payload")

	// ErrNoKubeClient is returned (and acked to the contract) when a
	// command needs Kubernetes but the provider has no kube client
	// wired. The text is part of the set_lease_env ack wire format —
	// do not change it.
	ErrNoKubeClient = errors.New("provider has no kube client wired")

	// ErrNoResponder means the configured ResponseMode requires the
	// x/orchestration submission path but no Responder was configured.
	ErrNoResponder = errors.New("orchestration response mode enabled but no Responder configured")

	// ErrUnknownResponseMode marks an unrecognized ResponseMode value.
	ErrUnknownResponseMode = errors.New("unknown orchestration response mode")
)

Sentinel errors for the provider orchestration package. Guard-style failures wrap these with fmt.Errorf("...: %w", Err...) so callers can branch with errors.Is instead of matching message strings.

Chain-side (x/orchestration) failures use registered Cosmos module errors instead — see node/x/orchestration/types/errors.go. The conventions are documented in docs/CONVENTIONS.md.

Functions

This section is empty.

Types

type Config

type Config struct {
	// PollInterval between blockchain polls
	PollInterval time.Duration `json:"poll_interval" yaml:"poll_interval"`
	// ProviderAddress is this provider's blockchain address
	ProviderAddress string `json:"provider_address" yaml:"provider_address"`
	// NodeRPCURL is the blockchain node RPC URL
	NodeRPCURL string `json:"node_rpc_url" yaml:"node_rpc_url"`
	// Enabled controls whether polling is active
	Enabled bool `json:"enabled" yaml:"enabled"`
	// ResponseMode chooses how responses are submitted back to the chain.
	// Defaults to contract-execute for backward compatibility.
	ResponseMode ResponseMode `json:"response_mode" yaml:"response_mode"`
	// KubeSettings is required with ClusterRead for NodePort discovery on
	// health/status responses (ClusterPublicHostname, etc.). Both may be nil.
	KubeSettings *builder.Settings `json:"-" yaml:"-"`
	// ClusterRead supplies ForwardedPortStatus when set (typically the real
	// cluster client from provider run).
	ClusterRead cluster.ReadClient `json:"-" yaml:"-"`
	// Deployments enumerates the leases the poller should proactively report
	// forwarded_ports for. When nil the poller falls back to the legacy
	// behaviour of only emitting ForwardedPorts in response to a chain
	// `health_request` event.
	Deployments DeploymentLister `json:"-" yaml:"-"`
	// ForwardedPortsInterval throttles how often the poller publishes a
	// fresh forwarded_ports snapshot for each active lease. Zero defaults
	// to one minute. The chain only keeps the latest payload per lease so
	// the interval can be conservative without losing fidelity.
	ForwardedPortsInterval time.Duration `json:"forwarded_ports_interval" yaml:"forwarded_ports_interval"`
	// ForwardedPortsOnChange switches the publisher to CHANGE-DRIVEN: a lease's
	// deployment_ready/forwarded_ports tx is submitted as soon as the cluster's
	// NodePort allocation DIFFERS from the value this process last published for
	// that lease (or has no cached value yet — the first deploy, and once after a
	// provider restart: the "after reset" case), instead of on every fixed tick.
	// Default false preserves the legacy timer behaviour.
	//
	// IMPORTANT: this is NOT "publish only on change and never again". An
	// unchanged lease is still republished every LivenessRefreshInterval (the
	// heartbeat backstop) so the per-lease liveness signal stays fresh and a dead
	// deployment on an otherwise-healthy provider is still detected + failed over.
	// Pure on-change publishing (no backstop) drops the heartbeat entirely, and a
	// dead pod on a provider whose gateway is up becomes invisible to the HA
	// contract (per-provider x/valiporacle can't see it). See
	// docs/legacy/LIVENESS-AND-ENDPOINT-REDESIGN.md §8.
	ForwardedPortsOnChange bool `json:"forwarded_ports_on_change" yaml:"forwarded_ports_on_change"`
	// LivenessRefreshInterval is the heartbeat backstop for change-driven mode:
	// an unchanged lease is republished at least this often. Zero inherits
	// ForwardedPortsInterval (resolved in Normalized — same liveness cadence as
	// the legacy timer, so enabling change-driven is safe by default). RAISE it —
	// with x/valiporacle handling fast provider-down failover — to cut
	// steady-state tx cost, at the price of slower per-deployment
	// (dead-pod-on-healthy-provider) failover. Must stay below the HA contract's
	// health_timeout (in wall-clock) or healthy members will spuriously time out.
	// Ignored when ForwardedPortsOnChange=false.
	LivenessRefreshInterval time.Duration `json:"liveness_refresh_interval" yaml:"liveness_refresh_interval"`
}

Config holds poller configuration

func DefaultConfig

func DefaultConfig() Config

DefaultConfig returns sensible defaults

func (Config) Normalized

func (c Config) Normalized() Config

Normalized returns a copy of the config with every unset value filled from DefaultConfig. It is the single defaulting point for this package: NewPollerWithResponder normalizes its input, so zero values coming from unset CLI flags / env vars degrade to the documented defaults instead of arming time.NewTicker with a non-positive interval (which panics). Enabled is never touched — a zero Config stays disabled.

func (Config) Validate

func (c Config) Validate() error

Validate reports configuration errors that should stop startup rather than silently degrade. Call it on the normalized config.

type DeploymentLister

type DeploymentLister interface {
	Deployments(ctx context.Context) ([]ctypes.IDeployment, error)
}

DeploymentLister enumerates the active leases this provider is serving so the poller can proactively publish each lease's NodePort/forwarded_ports to the chain's orchestration message queue, even when nothing has asked for a health check. Without this, the chain's wasm `active_provider_endpoints` query would only have a populated `workload_forwarded_ports` array when a contract (or operator) had just emitted a `request_health` event — the HA orchestrator contract has no such path, so the tenant container port was invisible in the routing response despite the workload actually running on a NodePort service.

`provider/cluster.Client` satisfies this interface; we keep the dependency minimal so the poller package does not need the full read/write surface and existing tests that inject a `cluster.ReadClient` stay valid.

type Handlers

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

Handlers is the orchestration business-logic service: it turns orchestration request messages into Kubernetes operations and pushes the results into the ResponseSink. It knows nothing about polling cadence or chain transport, so it can be unit-tested with fakes for every dependency.

func NewHandlers

func NewHandlers(
	logger log.Logger,
	healthChecker LeaseHealthChecker,
	kube KubeClient,
	clusterRead cluster.ReadClient,
	kubeSettings *builder.Settings,
	sink ResponseSink,
) *Handlers

NewHandlers constructs the business-logic service. health, kube, clusterRead and kubeSettings may be nil; the corresponding handlers degrade the same way the pre-service poller did (error acks or empty forwarded-port lists).

func (*Handlers) Process

Process handles a single orchestration message

func (*Handlers) SendHealthResponse

func (h *Handlers) SendHealthResponse(ctx context.Context, msg apiorch.OrchestrationMessage, healthResult *health.LeaseHealth) error

SendHealthResponse sends a health check response to a contract

type HealthResponsePayload

type HealthResponsePayload struct {
	LeaseID   apiorch.LeaseIdentifier `json:"lease_id"`
	Status    string                  `json:"status"`
	Message   string                  `json:"message,omitempty"`
	Timestamp int64                   `json:"timestamp"`
	// NodePort / workload-facing ports (from cluster ForwardedPortStatus), when configured.
	ForwardedPorts []orchestrationForwardedPort `json:"forwarded_ports,omitempty"`
}

HealthResponsePayload is the provider-dialect health/liveness payload (see the package comment above; do not confuse with the chain-canonical api/orchestration.HealthResponsePayload).

type KubeClient

type KubeClient interface {
	ScaleDeployment(ctx context.Context, namespace, service string, replicas int32) error
	RestartDeployment(ctx context.Context, namespace, service string) error
	StopDeployment(ctx context.Context, namespace, service string) error
	// SetLeaseEnv persists per-lease env vars in a ConfigMap inside
	// the lease's namespace (typically named
	// `akash-lease-env-<dseq>-<gseq>-<oseq>`). The builder reads this
	// ConfigMap when (re)creating pods so the env merges in
	// transparently. `version` is monotonically increasing; the
	// implementation must reject stale writes (version <= currently
	// stored version) and return the actually-applied version. When
	// `restart` is true the implementation also triggers a rolling
	// restart of every Deployment in the namespace so the env takes
	// effect immediately. Returns the ConfigMap name, the applied
	// version, whether a restart was triggered, and any error.
	SetLeaseEnv(
		ctx context.Context,
		namespace string,
		env map[string]string,
		version uint64,
		restart bool,
	) (configMapName string, applied uint64, restartTriggered bool, err error)
}

KubeClient provides Kubernetes operations for orchestration commands

type LeaseHealthChecker

type LeaseHealthChecker interface {
	CheckLeaseHealth(ctx context.Context, leaseID mv1.LeaseID) (*health.LeaseHealth, error)
}

LeaseHealthChecker abstracts health.KubeHealthChecker so the handler service can be exercised without a Kubernetes cluster.

type MessageSource

type MessageSource interface {
	PendingMessages(ctx context.Context) ([]apiorch.OrchestrationMessage, error)
}

MessageSource is the repository abstraction over wherever pending orchestration messages come from. Production uses the chain-backed implementation in chain_query.go (custom ABCI query with wasm-store and event fallbacks); tests inject a fake.

type Poller

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

Poller drives the orchestration request/response cycle: it reads pending messages from a MessageSource on a fixed cadence, dedupes them, and hands each one to the Handlers service. The business logic (Handlers) and the chain I/O (chainMessageSource / chainResponseSink) are injected, so the poller itself is only lifecycle + scheduling.

func NewPoller

func NewPoller(
	cfg Config,
	logger log.Logger,
	txClient client.Context,
	healthChecker *health.KubeHealthChecker,
	kubeClient KubeClient,
) *Poller

NewPoller creates a new orchestration message poller.

When cfg.ResponseMode is "orchestration" or "both", a Responder is instantiated automatically from the supplied txClient; callers that want custom gas/broadcast tuning can use NewPollerWithResponder.

func NewPollerWithResponder

func NewPollerWithResponder(
	cfg Config,
	logger log.Logger,
	txClient client.Context,
	healthChecker *health.KubeHealthChecker,
	kubeClient KubeClient,
	responder *Responder,
) *Poller

NewPollerWithResponder is like NewPoller but lets callers inject a pre-configured Responder (useful for tests and for tuning gas/broadcast mode differently from the poller's defaults).

func (*Poller) Handlers

func (p *Poller) Handlers() *Handlers

Handlers exposes the business-logic service backing this poller, so callers (and tests) can drive individual message handling without the polling loop.

func (*Poller) SendHealthResponse

func (p *Poller) SendHealthResponse(ctx context.Context, msg apiorch.OrchestrationMessage, healthResult *health.LeaseHealth) error

SendHealthResponse sends a health check response to a contract

func (*Poller) Start

func (p *Poller) Start(ctx context.Context) error

Start begins polling for messages

func (*Poller) Stop

func (p *Poller) Stop()

Stop stops the poller

type Responder

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

Responder submits MsgSubmitOrchestrationResponse transactions.

A single Responder is safe for concurrent use: each Submit call builds its own tx factory and fetches a fresh account/sequence, so nothing mutable leaks between callers. The underlying client.Context is treated as read-only.

func NewResponder

func NewResponder(cfg ResponderConfig, logger log.Logger, txClient client.Context) *Responder

NewResponder constructs a Responder. The passed client.Context must already have the orchestration codec types registered (done at app init via orchestrationtypes.RegisterInterfaces).

func (*Responder) Submit

func (r *Responder) Submit(ctx context.Context, opts SubmitOptions) (*SubmitResult, error)

Submit builds, signs, and broadcasts a MsgSubmitOrchestrationResponse.

The payload is auto-marshaled if it is not already a []byte or json.RawMessage. Nil payload is allowed for message types where the fact of responding is the signal (e.g. command_ack with no detail).

type ResponderConfig

type ResponderConfig struct {
	// ProviderAddress is this provider's bech32 signer address. Must
	// match the LeaseID.Provider of every message being submitted.
	ProviderAddress string

	// FromName is the keyring entry used to sign transactions. Usually
	// the same key used by the rest of the provider daemon.
	FromName string

	// Gas settings applied to every submission. Override for load
	// testing or when running on chains with unusual gas markets.
	Gas           uint64
	GasAdjustment float64
	GasPrices     string

	// BroadcastMode: "sync" (default), "async", or "block".
	BroadcastMode string

	// DefaultTTLBlocks is copied onto outbound messages when the caller
	// does not specify one. Zero means "let the chain apply its module
	// default" (see orchestration.Params.DefaultTTLBlocks).
	DefaultTTLBlocks uint64
}

ResponderConfig tunes the Responder. Zero values fall back to reasonable defaults appropriate for a provider daemon that wants to get responses on-chain quickly without blocking on finality.

func DefaultResponderConfig

func DefaultResponderConfig() ResponderConfig

DefaultResponderConfig returns tuned defaults matching the legacy sendResponse path so drop-in replacement is behavior-preserving.

Gas note: deployment_ready payloads carry forwarded_ports + per-service metadata and an `out of gas in location: WritePerByte` at ~1.22M gas was observed on multi-node lab providers when each lease has a single forwarded NodePort (payload_bytes=343). The keeper's persistence path is dominated by per-byte costs for the entire orchestration-state dump (every existing message gets re-serialised on update). Bumping to 2.5M leaves plenty of headroom for multi-port leases AND for the state-dump growing as more leases accumulate during a demo run.

type ResponseMode

type ResponseMode string

ResponseMode selects how the poller pushes responses back to the chain.

  • ResponseModeContract: legacy path, wraps responses in MsgExecuteContract addressed to the originating contract. Required for existing CosmWasm contracts that implement receive_orchestration_response.
  • ResponseModeOrchestration: new path (P0.1), submits MsgSubmitOrchestrationResponse directly to the x/orchestration module. The chain-side keeper queues the message for contract consumption and (for health_response) mirrors into x/market for backward-compat reads.
  • ResponseModeBoth: mirror-writes on both paths during migration so we can cut over contract-side without a coordinated deploy.
const (
	ResponseModeContract      ResponseMode = "contract"
	ResponseModeOrchestration ResponseMode = "orchestration"
	ResponseModeBoth          ResponseMode = "both"
)

func ParseResponseMode

func ParseResponseMode(s string) (ResponseMode, error)

ParseResponseMode normalizes a CLI/config value into a ResponseMode. Empty string returns the legacy default to preserve existing behavior.

type ResponseSink

type ResponseSink interface {
	SendResponse(ctx context.Context, origin apiorch.OrchestrationMessage, msgType apiorch.OrchestrationMessageType, payload interface{}) error
}

ResponseSink abstracts how handler results are delivered back to the chain. Production uses the chain-backed implementation in response.go (contract-execute and/or x/orchestration submission per ResponseMode); tests inject a fake and assert on the captured responses.

type SubmitOptions

type SubmitOptions struct {
	LeaseID       apiorch.LeaseIdentifier
	MessageType   string
	CorrelationID string
	// Payload is an arbitrary JSON-encodable value. Pass []byte / json.RawMessage
	// to skip re-encoding, or a Go struct to let Submit marshal it.
	Payload   interface{}
	TTLBlocks uint64
}

SubmitOptions configure a single Submit call. The Responder fills in defaults from its config for anything left zero.

type SubmitResult

type SubmitResult struct {
	TxHash  string `json:"tx_hash"`
	Code    uint32 `json:"code"`
	RawLog  string `json:"raw_log"`
	GasUsed int64  `json:"gas_used"`
	// MessageID is the queue id returned by the chain in the tx response
	// events. Empty when the broadcast mode does not include a result
	// (e.g. async) or when the chain did not emit the expected event.
	MessageID string `json:"message_id"`
}

SubmitResult summarizes the outcome of a successful broadc ast.

JSON tags use snake_case so script consumers (e.g. scripts/single-node/test-orchestration-response.sh) can grep on the canonical Cosmos field names without depending on Go's exported-field casing.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL