provisioner

package
v0.10.0 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: 27 Imported by: 0

Documentation

Overview

Package provisioner implements the lease provisioning lifecycle using Watermill for event-driven processing.

Overview

The provisioner bridges chain events to backend operations: 

Chain Event → Watermill Topic → Handler → Backend Call → Chain Tx

Manager

The Manager is the central component that:

  • Subscribes to Watermill topics for lease events
  • Tracks in-flight provisions to prevent duplicates
  • Routes to appropriate backends via the backend.Router
  • Batches lease acknowledgments for efficiency
  • Handles backend callbacks and chain transactions

Event Topics

The provisioner uses these Watermill topics: 

TopicLeaseCreated     - New lease needs provisioning
TopicLeaseClosed      - Lease closed, deprovision resources
TopicLeaseExpired     - Lease expired, deprovision resources
TopicPayloadReceived  - Tenant uploaded payload, start provisioning
TopicBackendCallback  - Backend reported provisioning result
TopicLeaseEvent       - Real-time lease status events for WebSocket delivery

Reconciler

The Reconciler provides level-triggered state reconciliation:

  • Runs on startup to recover from crashes
  • Runs periodically to fix any drift
  • Compares chain state vs backend state
  • Takes corrective action (provision, deprovision, acknowledge)

PayloadStore

The PayloadStore provides temporary storage for tenant deployment payloads:

  • Stores payloads uploaded via API until provisioning starts
  • Uses write batching for efficiency under load
  • Reconciler-driven cleanup when leases are no longer active
  • Survives restarts (persistent bbolt storage)

Crash Recovery

The provisioner uses level-triggered reconciliation for crash recovery. Rather than replaying missed events, it queries current state and acts: 

Chain: PENDING lease exists
Backend: No provision found
Action: Start provisioning

This approach is simpler and handles any inconsistency, not just missed events.

Index

Constants

View Source 
const (
	// DefaultAckBatchInterval is the maximum time to wait before flushing a batch.
	// Short interval to minimize latency while still allowing batching.
	DefaultAckBatchInterval = 500 * time.Millisecond

	// DefaultAckBatchSize is the maximum number of acks to batch before flushing.
	// With authz sub-signers, each lane can handle a large batch per block.
	DefaultAckBatchSize = 50
)
View Source 
const (
	TopicLeaseCreated    = "events.lease.created"
	TopicLeaseClosed     = "events.lease.closed"
	TopicLeaseExpired    = "events.lease.expired"
	TopicBackendCallback = "events.backend.callback"
	TopicPayloadReceived = "events.payload.received"
	TopicLeaseEvent      = "events.lease.event"
)

Watermill topic names for internal event routing.

View Source 
const CallbackPath = "/callbacks/provision"

CallbackPath is the path suffix for backend provision callbacks.

View Source 
const DefaultMaxReprovisionAttempts = 3

DefaultMaxReprovisionAttempts is the default number of re-provision attempts before rejecting a lease whose containers keep failing.

View Source 
const (
	// DefaultReconcileWorkers is the default number of concurrent workers for
	// processing leases and orphans during reconciliation.
	DefaultReconcileWorkers = 10
)

Default concurrency limits for reconciliation.

Variables

View Source 
var (
	// ErrMalformedMessage indicates the message payload could not be parsed.
	// This is a terminal error - the message should not be retried.
	ErrMalformedMessage = errors.New("malformed message payload")

	// ErrNoBackendAvailable indicates no backend is configured to handle the request.
	ErrNoBackendAvailable = errors.New("no backend available")

	// ErrProvisioningFailed indicates the backend failed to provision the resource.
	ErrProvisioningFailed = errors.New("provisioning failed")

	// ErrDeprovisionFailed indicates the backend failed to deprovision the resource.
	ErrDeprovisionFailed = errors.New("deprovision failed")

	// ErrAcknowledgeFailed indicates the lease acknowledgment on chain failed.
	ErrAcknowledgeFailed = errors.New("lease acknowledgment failed")
)

Sentinel errors for provisioner operations.

Functions

func BuildCallbackURL 

func BuildCallbackURL(baseURL string) string

BuildCallbackURL constructs the full callback URL from a base URL.

func ExtractLeaseItems 

func ExtractLeaseItems(lease *billingtypes.Lease) []backend.LeaseItem

ExtractLeaseItems converts chain lease items to backend lease items.

func ExtractRoutingSKU 

func ExtractRoutingSKU(lease *billingtypes.Lease) string

ExtractRoutingSKU returns a SKU UUID from the lease for backend routing.

Why this exists: A lease may contain multiple items with different SKUs, but all items are guaranteed to belong to the same provider (enforced by the chain). Therefore, any SKU can be used to determine which backend should handle the request. We use the first item's SKU by convention.

This should NOT be used for resource allocation - use ExtractLeaseItems() to get the full list of items with their quantities.

func TotalLeaseQuantity 

func TotalLeaseQuantity(lease *billingtypes.Lease) int

TotalLeaseQuantity returns the total quantity across all lease items.

Types

type AckBatcher 

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

AckBatcher distributes lease acknowledgment requests across N independent lanes. Each lane runs its own sequential batchLoop, enabling parallel broadcasting when backed by multiple signers via authz.

func NewAckBatcher 

func NewAckBatcher(chainClient ChainClient, cfg AckBatcherConfig) *AckBatcher

NewAckBatcher creates a new acknowledgment batcher with N lanes.

func (*AckBatcher) Acknowledge 

func (b *AckBatcher) Acknowledge(ctx context.Context, leaseUUID string) (bool, string, error)

Acknowledge queues a lease for acknowledgment and waits for the result. Requests are distributed across lanes via round-robin.

func (*AckBatcher) Start 

func (b *AckBatcher) Start(ctx context.Context)

Start begins the batching loop for all lanes.

func (*AckBatcher) Stop 

func (b *AckBatcher) Stop()

Stop gracefully shuts down all lanes, flushing pending requests.

type AckBatcherConfig 

type AckBatcherConfig struct {
	// ProviderUUID is required for querying pending leases.
	ProviderUUID string

	// BatchInterval is the maximum time to wait before flushing a batch.
	// Default: 500ms
	BatchInterval time.Duration

	// BatchSize is the maximum number of acks to collect before flushing.
	// Default: DefaultAckBatchSize (50)
	BatchSize int

	// LaneCount is the number of parallel lanes. Defaults to 1 (single signer).
	// With N sub-signers, set to N.
	LaneCount int
}

AckBatcherConfig configures the acknowledgment batcher.

type Acknowledger 

type Acknowledger interface {
	// Acknowledge queues a lease for acknowledgment and waits for the result.
	// Returns true if the lease is in an acknowledged state (either by this call
	// or already acknowledged), along with the transaction hash (empty if no tx
	// was needed). This method blocks until the acknowledgment is processed.
	Acknowledge(ctx context.Context, leaseUUID string) (acknowledged bool, txHash string, err error)
}

Acknowledger defines the interface for acknowledging leases on chain. This is used by handlers to acknowledge successful provisions.

type BackendRouter 

type BackendRouter interface {
	// Route returns the appropriate backend for the given SKU.
	Route(sku string) backend.Backend

	// RouteForProvision selects the least-loaded backend matching the SKU for a
	// new provision, falling back to round-robin when no candidate exposes usable
	// load stats. inFlightByBackend is a per-backend in-flight provision count
	// used to spread concurrent provisions; it may be nil.
	RouteForProvision(ctx context.Context, sku string, inFlightByBackend map[string]int) backend.Backend

	// GetBackendByName returns a backend by its name. Returns nil if not found.
	GetBackendByName(name string) backend.Backend

	// Backends returns all unique backends for operations like reconciliation.
	Backends() []backend.Backend
}

BackendRouter defines the interface for routing requests to backends. This abstracts the backend.Router for testability.

type ChainClient 

type ChainClient interface {
	GetLease(ctx context.Context, leaseUUID string) (*billingtypes.Lease, error)
	GetPendingLeases(ctx context.Context, providerUUID string) ([]billingtypes.Lease, error)
	AcknowledgeLeases(ctx context.Context, leaseUUIDs []string) (uint64, []string, error)
	RejectLeases(ctx context.Context, leaseUUIDs []string, reason string) (uint64, []string, error)
	CloseLeases(ctx context.Context, leaseUUIDs []string, reason string) (uint64, []string, error)
}

ChainClient defines the chain operations needed by the provisioner.

type DefaultInFlightTracker 

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

DefaultInFlightTracker is the default implementation of InFlightTracker. It uses a sync.RWMutex for thread-safe tracking of in-flight provisions.

func NewInFlightTracker 

func NewInFlightTracker() *DefaultInFlightTracker

NewInFlightTracker creates a new DefaultInFlightTracker.

func (*DefaultInFlightTracker) GetInFlight 

func (t *DefaultInFlightTracker) GetInFlight(leaseUUID string) (InFlightProvision, bool)

GetInFlight returns the in-flight provision info without removing it.

func (*DefaultInFlightTracker) GetInFlightLeases 

func (t *DefaultInFlightTracker) GetInFlightLeases() []string

GetInFlightLeases returns a snapshot of all in-flight lease UUIDs.

func (*DefaultInFlightTracker) GetTimedOutProvisions 

func (t *DefaultInFlightTracker) GetTimedOutProvisions(timeout time.Duration) []InFlightProvision

GetTimedOutProvisions returns provisions that have exceeded the given timeout.

func (*DefaultInFlightTracker) InFlightCount 

func (t *DefaultInFlightTracker) InFlightCount() int

InFlightCount returns the number of provisions currently in flight.

func (*DefaultInFlightTracker) InFlightCountsByBackend added in v0.5.0 

func (t *DefaultInFlightTracker) InFlightCountsByBackend() map[string]int

InFlightCountsByBackend returns a snapshot of in-flight provision counts keyed by backend name.

func (*DefaultInFlightTracker) IsInFlight 

func (t *DefaultInFlightTracker) IsInFlight(leaseUUID string) bool

IsInFlight checks if a lease is currently being provisioned.

func (*DefaultInFlightTracker) PopInFlight 

func (t *DefaultInFlightTracker) PopInFlight(leaseUUID string) (InFlightProvision, bool)

PopInFlight atomically removes and returns an in-flight provision.

func (*DefaultInFlightTracker) TrackInFlight 

func (t *DefaultInFlightTracker) TrackInFlight(leaseUUID, tenant string, items []backend.LeaseItem, backendName string)

TrackInFlight registers a lease as being provisioned.

func (*DefaultInFlightTracker) TrackInFlightWithStartTime 

func (t *DefaultInFlightTracker) TrackInFlightWithStartTime(leaseUUID, tenant string, items []backend.LeaseItem, backendName string, startTime time.Time)

TrackInFlightWithStartTime is a testing helper that allows setting a custom start time. This should only be used in tests to simulate old provisions for timeout testing.

func (*DefaultInFlightTracker) TryTrackInFlight 

func (t *DefaultInFlightTracker) TryTrackInFlight(leaseUUID, tenant string, items []backend.LeaseItem, backendName string) bool

TryTrackInFlight atomically checks if a lease is already in-flight and tracks it if not. Returns true if the lease was successfully tracked (was not already in-flight), false if the lease was already being provisioned.

func (*DefaultInFlightTracker) TryTrackRestoreInFlight added in v0.5.0 

func (t *DefaultInFlightTracker) TryTrackRestoreInFlight(leaseUUID, tenant string, items []backend.LeaseItem, backendName string) bool

TryTrackRestoreInFlight is TryTrackInFlight for a restore: it marks the entry as KindRestore so the provision callback is acknowledged inline (ENG-358) and its metrics carry operation=restore. Like TryTrackInFlight it is atomic and returns false (leaving any existing entry untouched) when the lease is already in-flight, so a duplicate restore or a racing reconciler is a no-op for the caller.

func (*DefaultInFlightTracker) UntrackInFlight 

func (t *DefaultInFlightTracker) UntrackInFlight(leaseUUID string)

UntrackInFlight removes a lease from the in-flight tracking.

func (*DefaultInFlightTracker) WaitForDrain 

func (t *DefaultInFlightTracker) WaitForDrain(ctx context.Context, timeout time.Duration) int

WaitForDrain waits for all in-flight provisions to complete, up to the given timeout. Returns the number of provisions that were still in-flight when the timeout expired.

type EventBridge 

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

EventBridge bridges chain events to Watermill via the Manager. It subscribes to the EventSubscriber and publishes to Watermill topics.

func NewEventBridge 

func NewEventBridge(subscriber *chain.EventSubscriber, publisher EventPublisher) *EventBridge

NewEventBridge creates a new event bridge.

func (*EventBridge) Ready 

func (b *EventBridge) Ready() <-chan struct{}

Ready returns a channel that is closed when the bridge has subscribed and is ready to forward events.

func (*EventBridge) Start 

func (b *EventBridge) Start(ctx context.Context) error

Start begins forwarding events from the chain subscriber to Watermill. It must be called exactly once and should be run in a goroutine.

type EventPublisher 

type EventPublisher interface {
	PublishLeaseEvent(event chain.LeaseEvent) error
}

EventPublisher publishes chain events to Watermill.

type HandlerDeps 

type HandlerDeps struct {
	ChainClient   ChainClient
	Orchestrator  *ProvisionOrchestrator
	Tracker       InFlightTracker
	Acknowledger  Acknowledger
	PayloadStore  *payload.Store
	Publisher     message.Publisher // For publishing to TopicLeaseEvent (optional)
	BackendRouter BackendRouter     // Used to allowlist backend names on non-in-flight callback metrics
}

HandlerDeps contains the dependencies needed by the handler set.

type HandlerSet 

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

HandlerSet contains the Watermill message handlers for the provisioner. It encapsulates all handler methods and their dependencies.

func NewHandlerSet 

func NewHandlerSet(deps HandlerDeps) *HandlerSet

NewHandlerSet creates a new HandlerSet with the given dependencies.

func (*HandlerSet) HandleBackendCallback 

func (h *HandlerSet) HandleBackendCallback(msg *message.Message) (err error)

HandleBackendCallback processes callbacks from backends.

func (*HandlerSet) HandleLeaseClosed 

func (h *HandlerSet) HandleLeaseClosed(msg *message.Message) (err error)

HandleLeaseClosed processes lease closure events.

func (*HandlerSet) HandleLeaseCreated 

func (h *HandlerSet) HandleLeaseCreated(msg *message.Message) (err error)

HandleLeaseCreated processes new lease events.

func (*HandlerSet) HandleLeaseExpired 

func (h *HandlerSet) HandleLeaseExpired(msg *message.Message) (err error)

HandleLeaseExpired processes lease expiration events. Same logic as HandleLeaseClosed but records metrics under the correct topic.

func (*HandlerSet) HandlePayloadReceived 

func (h *HandlerSet) HandlePayloadReceived(msg *message.Message) (err error)

HandlePayloadReceived processes payload upload events. This triggers provisioning for leases that were waiting for a payload.

type InFlightProvision 

type InFlightProvision struct {
	LeaseUUID string
	Tenant    string
	Items     []backend.LeaseItem // All items being provisioned
	Backend   string
	StartTime time.Time     // For duration metrics
	Kind      ProvisionKind // Provision vs restore — labels callback metrics (ENG-358)
}

InFlightProvision represents a lease that is currently being provisioned.

func (InFlightProvision) RoutingSKU 

func (p InFlightProvision) RoutingSKU() string

RoutingSKU returns the first SKU for backend routing decisions.

Used during deprovision when we need to determine which backend handled a lease but only have the in-flight tracking data. Since all items in a lease belong to the same provider, any SKU works for routing.

This should NOT be used for resource calculations - use Items directly.

type InFlightTracker 

type InFlightTracker interface {
	// TryTrackInFlight atomically checks if a lease is already in-flight and tracks it if not.
	// Returns true if the lease was successfully tracked (was not already in-flight),
	// false if the lease was already being provisioned.
	TryTrackInFlight(leaseUUID, tenant string, items []backend.LeaseItem, backendName string) bool

	// TryTrackRestoreInFlight is TryTrackInFlight for a restore: the entry is
	// marked KindRestore so the restore's provision callback is acknowledged
	// inline rather than by the reconciler, and its metrics carry operation=restore (ENG-358).
	TryTrackRestoreInFlight(leaseUUID, tenant string, items []backend.LeaseItem, backendName string) bool

	// TrackInFlight registers a lease as being provisioned.
	TrackInFlight(leaseUUID, tenant string, items []backend.LeaseItem, backendName string)

	// UntrackInFlight removes a lease from the in-flight tracking.
	UntrackInFlight(leaseUUID string)

	// PopInFlight atomically removes and returns an in-flight provision.
	// Returns the provision info and true if found, or zero value and false if not found.
	PopInFlight(leaseUUID string) (InFlightProvision, bool)

	// GetInFlight returns the in-flight provision info without removing it.
	// Returns the provision info and true if found, or zero value and false if not found.
	GetInFlight(leaseUUID string) (InFlightProvision, bool)

	// IsInFlight checks if a lease is currently being provisioned.
	IsInFlight(leaseUUID string) bool

	// InFlightCount returns the number of provisions currently in flight.
	InFlightCount() int

	// InFlightCountsByBackend returns a snapshot of the number of in-flight
	// provisions per backend name. Used by the router's burst guard to spread
	// concurrent provisions that observe an identical backend load snapshot.
	InFlightCountsByBackend() map[string]int

	// GetInFlightLeases returns a snapshot of all in-flight lease UUIDs.
	GetInFlightLeases() []string

	// WaitForDrain waits for all in-flight provisions to complete, up to the given timeout.
	// Returns the number of provisions that were still in-flight when the timeout expired.
	WaitForDrain(ctx context.Context, timeout time.Duration) int

	// GetTimedOutProvisions returns provisions that have exceeded the given timeout.
	GetTimedOutProvisions(timeout time.Duration) []InFlightProvision
}

InFlightTracker defines the interface for tracking in-flight provisions. This is used by handlers, orchestrator, timeout checker, and reconciler.

type LeaseEventSink 

type LeaseEventSink interface {
	Publish(event backend.LeaseStatusEvent)
}

LeaseEventSink receives lease status events for real-time delivery (e.g., WebSocket).

type LeaseRejecter 

type LeaseRejecter interface {
	// RejectLeases rejects the given leases with the specified reason.
	// Returns the number of leases rejected, transaction hashes, and any error.
	RejectLeases(ctx context.Context, leaseUUIDs []string, reason string) (uint64, []string, error)
}

LeaseRejecter defines the interface for rejecting leases on chain. This is used by the TimeoutChecker to reject timed-out leases.

type Manager 

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

Manager handles the provisioning lifecycle using Watermill for event routing.

func NewManager 

func NewManager(cfg ManagerConfig, router *backend.Router, chainClient ChainClient) (*Manager, error)

NewManager creates a new provision manager with Watermill routing.

func (*Manager) AckBatcher 

func (m *Manager) AckBatcher() Acknowledger

AckBatcher returns the batcher as an Acknowledger for use by the reconciler.

func (*Manager) Close 

func (m *Manager) Close() error

Close shuts down the provision manager.

func (*Manager) DeletePayload 

func (m *Manager) DeletePayload(leaseUUID string)

DeletePayload removes a payload from the store. Used for rollback when publish fails after store succeeds. No-op if the payload store is not configured.

func (*Manager) GetInFlight 

func (m *Manager) GetInFlight(leaseUUID string) (InFlightProvision, bool)

GetInFlight delegates to the tracker.

func (*Manager) GetInFlightLeases 

func (m *Manager) GetInFlightLeases() []string

GetInFlightLeases delegates to the tracker.

func (*Manager) GetTimedOutProvisions 

func (m *Manager) GetTimedOutProvisions(timeout time.Duration) []InFlightProvision

GetTimedOutProvisions delegates to the tracker.

func (*Manager) HasPayload 

func (m *Manager) HasPayload(leaseUUID string) (bool, error)

HasPayload checks if a payload exists for a lease. Returns false if the payload store is not configured.

func (*Manager) InFlightCount 

func (m *Manager) InFlightCount() int

InFlightCount delegates to the tracker.

func (*Manager) InFlightCountsByBackend added in v0.5.0 

func (m *Manager) InFlightCountsByBackend() map[string]int

InFlightCountsByBackend delegates to the tracker.

func (*Manager) IsInFlight 

func (m *Manager) IsInFlight(leaseUUID string) bool

IsInFlight delegates to the tracker.

func (*Manager) PayloadStore 

func (m *Manager) PayloadStore() *payload.Store

PayloadStore returns the payload store for reconciliation access. May return nil if payload store is not configured.

func (*Manager) PopInFlight 

func (m *Manager) PopInFlight(leaseUUID string) (InFlightProvision, bool)

PopInFlight delegates to the tracker.

func (*Manager) PublishCallback 

func (m *Manager) PublishCallback(callback backend.CallbackPayload) error

PublishCallback publishes a backend callback to Watermill. This is called by the API server when it receives a callback.

func (*Manager) PublishLeaseEvent 

func (m *Manager) PublishLeaseEvent(event chain.LeaseEvent) error

PublishLeaseEvent publishes a chain event to the appropriate Watermill topic. This is called by the chain event subscriber.

func (*Manager) PublishPayload 

func (m *Manager) PublishPayload(event payload.Event) error

PublishPayload publishes a payload received event to Watermill. This is called by the API server when it receives a valid payload upload.

func (*Manager) RecordRestorePlacement added in v0.5.0 

func (m *Manager) RecordRestorePlacement(newLeaseUUID, backendName string)

RecordRestorePlacement delegates to the orchestrator (ENG-333).

func (*Manager) Running 

func (m *Manager) Running() chan struct{}

Running returns a channel that is closed when the router is running. This can be used to wait for the manager to be ready before publishing events.

func (*Manager) Start 

func (m *Manager) Start(ctx context.Context) error

Start begins the Watermill router and callback timeout checker.

func (*Manager) StorePayload 

func (m *Manager) StorePayload(leaseUUID string, payload []byte) bool

StorePayload stores a payload in the payload store. Returns false if a payload already exists for this lease (conflict), or if the payload store is not configured.

func (*Manager) TimeoutChecker 

func (m *Manager) TimeoutChecker() *TimeoutChecker

TimeoutChecker returns the internal timeout checker for testing purposes. This should only be used in tests.

func (*Manager) TrackInFlight 

func (m *Manager) TrackInFlight(leaseUUID, tenant string, items []backend.LeaseItem, backendName string)

TrackInFlight delegates to the tracker.

func (*Manager) Tracker 

func (m *Manager) Tracker() InFlightTracker

Tracker returns the internal tracker for testing purposes. This should only be used in tests.

func (*Manager) TryTrackInFlight 

func (m *Manager) TryTrackInFlight(leaseUUID, tenant string, items []backend.LeaseItem, backendName string) bool

TryTrackInFlight delegates to the tracker.

func (*Manager) TryTrackRestoreInFlight added in v0.5.0 

func (m *Manager) TryTrackRestoreInFlight(leaseUUID, tenant string, items []backend.LeaseItem, backendName string) bool

TryTrackRestoreInFlight delegates to the tracker. It lets the API restore handler register the new lease in-flight (as a restore) so the restore's provision callback is acknowledged inline rather than ~one reconciler interval later (ENG-358).

func (*Manager) UntrackInFlight 

func (m *Manager) UntrackInFlight(leaseUUID string)

UntrackInFlight delegates to the tracker.

func (*Manager) WaitForDrain 

func (m *Manager) WaitForDrain(ctx context.Context, timeout time.Duration) int

WaitForDrain delegates to the tracker.

type ManagerConfig 

type ManagerConfig struct {
	ProviderUUID         string
	CallbackBaseURL      string         // Base URL for backend callbacks (e.g., "http://fred.example.com:8080")
	PayloadStore         *payload.Store // Optional external payload store (if nil, manager won't handle payloads)
	PlacementStore       PlacementStore // Optional placement store for round-robin routing (nil = disabled)
	LeaseEventSink       LeaseEventSink // Optional sink for real-time lease events (nil = disabled)
	CallbackTimeout      time.Duration  // Timeout for backend callbacks (default: 10 minutes, 0 = disabled)
	TimeoutCheckInterval time.Duration  // How often to check for timeouts (default: 1 minute)
	AckBatchInterval     time.Duration  // How long to wait before flushing ack batch (default: DefaultAckBatchInterval)
	AckBatchSize         int            // Maximum acks to batch before flushing (default: DefaultAckBatchSize)
	AckLaneCount         int            // Number of parallel ack lanes (default: 1)
}

ManagerConfig configures the provision manager.

type PlacementStore 

type PlacementStore interface {
	Get(leaseUUID string) string
	SetAt(leaseUUID string) (time.Time, bool)
	Set(leaseUUID, backendName string) error
	Delete(leaseUUID string)
	SetBatch(placements map[string]string) error
	Count() int
	List() []string
	Healthy() error
	Close() error
}

PlacementStore records which backend is serving each lease so that read operations reach the correct backend after provision routing.

type ProvisionKind added in v0.5.0 

type ProvisionKind uint8

ProvisionKind distinguishes a fresh provision from a restore. A restore IS a provisioning operation — it brings a lease's deployment up from retained volumes instead of a fresh manifest — so it shares the provisioning metrics and is differentiated by an operation label rather than a separate metric, per Prometheus naming guidance (sum/avg across the dimension stays meaningful). ENG-358 tracks restores in-flight so their callback is acknowledged inline; the Kind keeps restore latency/outcomes labeled distinctly from fresh provisions (which ENG-357 deliberately kept separable). 

const (
	KindProvision ProvisionKind = iota // zero value: a fresh provision
	KindRestore                        // a restore from retained volumes (ENG-358)
)

type ProvisionOpts 

type ProvisionOpts struct {
	Payload     []byte // Optional deployment payload
	PayloadHash string // Optional hex-encoded SHA-256 hash of payload
}

ProvisionOpts contains optional parameters for provisioning.

type ProvisionOrchestrator 

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

ProvisionOrchestrator coordinates the provisioning flow. It routes to the appropriate backend, tracks the provision in-flight, and initiates the async provisioning call.

func NewProvisionOrchestrator 

func NewProvisionOrchestrator(providerUUID, callbackBaseURL string, router BackendRouter, tracker InFlightTracker, placementStore PlacementStore) *ProvisionOrchestrator

NewProvisionOrchestrator creates a new ProvisionOrchestrator.

func (*ProvisionOrchestrator) DeletePlacement 

func (o *ProvisionOrchestrator) DeletePlacement(leaseUUID string)

DeletePlacement removes the placement record for a lease. Called when a lease reaches a terminal state (e.g., rejected after a failure callback) without going through the full Deprovision flow.

func (*ProvisionOrchestrator) Deprovision 

func (o *ProvisionOrchestrator) Deprovision(ctx context.Context, leaseUUID string) error

Deprovision tears down a lease's backend resources. The backend is resolved POSITIVELY — from the placement record, then the in-flight tracker. It never guesses a default backend from the SKU: in a multi-backend pool a SKU is not pinned to one backend, so a guessed deprovision is a phantom no-op that reports success while stranding the real volume on another backend (ENG-335). When the backend cannot be positively resolved, all backends are swept; deprovision is idempotent, so the real holder is torn down and the rest are harmless no-ops.

Returns nil on success or if the lease was not provisioned anywhere. Returns an error only if every attempted deprovision failed.

func (*ProvisionOrchestrator) RecordRestorePlacement added in v0.5.0 

func (o *ProvisionOrchestrator) RecordRestorePlacement(newLeaseUUID, backendName string)

RecordRestorePlacement optimistically records the NEW lease's placement after a successful restore (the new lease now lives on the backend that held the source's retained data). No-op when no placement store is configured (nil interface). It deliberately does NOT delete the source placement: restore is asynchronous (202 + adopt), so deleting source state before the adopt confirms is a saga anti-pattern, and source-placement cleanup is owned solely by the reconciler (which prunes it once the retention disappears from /retentions). This just closes the post-restore reconcile window for the new lease (ENG-333).

func (*ProvisionOrchestrator) StartProvisioning 

func (o *ProvisionOrchestrator) StartProvisioning(ctx context.Context, lease *billingtypes.Lease, opts ProvisionOpts) error

StartProvisioning handles the common provisioning flow for both lease creation and payload-triggered provisioning. It routes to the appropriate backend, tracks the provision in-flight, and initiates the async provisioning call.

Returns nil if provisioning was started successfully or the lease is already in-flight. Returns an error if routing fails or the backend call fails.

type Reconciler 

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

Reconciler performs level-triggered reconciliation between chain state and backend state. It ensures consistency by comparing current state rather than replaying events.

func NewReconciler 

func NewReconciler(cfg ReconcilerConfig, chainClient ReconcilerChainClient, acknowledger Acknowledger, backendRouter BackendRouter, tracker ReconcilerTracker, placementStore PlacementStore) (*Reconciler, error)

NewReconciler creates a new reconciler. The acknowledger (required) routes ack operations through the batcher for parallel signing. The tracker parameter is optional - if nil, the reconciler will not coordinate with the event-driven path. The placementStore parameter is optional - if nil, placement tracking is disabled.

func (*Reconciler) ReconcileAll 

func (r *Reconciler) ReconcileAll(ctx context.Context) (retErr error)

ReconcileAll performs a full reconciliation between chain state and backend state. This is the core level-triggered reconciliation logic.

State Matrix: | Chain State | Backend State | Action | |-------------|---------------|--------| | PENDING | Not provisioned | Start provisioning | | PENDING | Provisioning (in progress) | Nothing (wait for callback) | | PENDING | Provisioned + ready (in-flight) | Skip; main flow owns the ack | | PENDING | Provisioned + ready | Acknowledge lease | | PENDING | Provisioned + failed | Reject lease on chain | | ACTIVE | Provisioned + ready | Nothing (healthy) | | ACTIVE | Provisioned + failed | Re-provision (close after max attempts) | | ACTIVE | Not provisioned | Anomaly: re-provision with payload | | Not found | Provisioned | Orphan: Deprovision |

func (*Reconciler) RunOnce 

func (r *Reconciler) RunOnce(ctx context.Context) error

RunOnce performs a single reconciliation. Use this at startup.

func (*Reconciler) Start 

func (r *Reconciler) Start(ctx context.Context) error

Start begins periodic reconciliation.

type ReconcilerChainClient 

type ReconcilerChainClient interface {
	GetPendingLeases(ctx context.Context, providerUUID string) ([]billingtypes.Lease, error)
	GetActiveLeasesByProvider(ctx context.Context, providerUUID string) ([]billingtypes.Lease, error)
	AcknowledgeLeases(ctx context.Context, leaseUUIDs []string) (uint64, []string, error)
	RejectLeases(ctx context.Context, leaseUUIDs []string, reason string) (uint64, []string, error)
	CloseLeases(ctx context.Context, leaseUUIDs []string, reason string) (uint64, []string, error)
}

ReconcilerChainClient defines the chain operations needed by the reconciler.

type ReconcilerConfig 

type ReconcilerConfig struct {
	ProviderUUID           string
	CallbackBaseURL        string
	Interval               time.Duration // How often to run periodic reconciliation
	MaxWorkers             int           // Maximum concurrent workers (default: 10)
	MaxReprovisionAttempts int           // Max re-provision attempts before rejecting (default: 3)
}

ReconcilerConfig configures the reconciler.

type ReconcilerTracker 

type ReconcilerTracker interface {
	InFlightTracker

	// HasPayload checks if a payload exists for a lease.
	// Returns an error if the underlying store read fails.
	HasPayload(leaseUUID string) (bool, error)

	// PayloadStore returns the payload store for direct access.
	// May return nil if payload store is not configured.
	PayloadStore() *payload.Store
}

ReconcilerTracker extends InFlightTracker with payload-related methods needed by the reconciler for coordinating with the event-driven path.

type TimeoutChecker 

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

TimeoutChecker monitors in-flight provisions and rejects timed-out ones. It runs as a background goroutine and periodically checks for provisions that have exceeded the callback timeout.

func NewTimeoutChecker 

func NewTimeoutChecker(cfg TimeoutCheckerConfig) *TimeoutChecker

NewTimeoutChecker creates a new TimeoutChecker.

func (*TimeoutChecker) CheckOnce 

func (c *TimeoutChecker) CheckOnce(ctx context.Context)

CheckOnce performs a single timeout check. This is exposed for testing.

func (*TimeoutChecker) Start 

func (c *TimeoutChecker) Start(ctx context.Context)

Start begins the timeout checker loop. It runs until the context is canceled.

type TimeoutCheckerConfig 

type TimeoutCheckerConfig struct {
	Tracker       InFlightTracker
	Rejecter      LeaseRejecter
	Timeout       time.Duration // Callback timeout (how long to wait before considering a provision timed out)
	CheckInterval time.Duration // How often to check for timeouts
}

TimeoutCheckerConfig configures the timeout checker.

Source Files

View all Source files

Directories

Path Synopsis
Package payload implements bbolt-backed temporary storage for tenant deployment payloads.
 Package payload implements bbolt-backed temporary storage for tenant deployment payloads.
Package placement implements a bbolt-backed lease→backend mapping with an in-memory cache.
 Package placement implements a bbolt-backed lease→backend mapping with an in-memory cache.

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.