orchestrator

package
v0.3.0-20260805154417-... Latest Latest
Warning

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

 Go to latest Published: Aug 5, 2026 License: Apache-2.0 Imports: 25 Imported by: 0

README

SubmitQueue Orchestrator

Documentation

Overview

Package orchestrator declares the SubmitQueue orchestrator's pipeline topology, extension seams, and controller set. The host (main.go) fills Deps and passes Stages to pipeline.Construct; no assembly logic lives here.

Index

Constants

This section is empty.

Variables

View Source 
var PublishOnlyTopics = []pipeline.PublishOnlyTopic{

	{Key: topickey.TopicKeyLog, Name: "log"},

	{Key: runwaymq.TopicKeyMergeConflictCheck, Name: "merge-conflict-check"},

	{Key: runwaymq.TopicKeyMerge, Name: "runway-merge"},
}

PublishOnlyTopics declares topics the orchestrator publishes to but does not consume. These are registered in the TopicRegistry so controllers can look up topic names for publishing.

View Source 
var Stages = []pipeline.Stage[Deps]{
	{
		Key:           topickey.TopicKeyStart,
		Name:          "start",
		ConsumerGroup: "orchestrator",
		New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
			return start.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
		},
		DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
			return dlq.NewDLQRequestController(d.Logger, d.Scope, d.Storage, sc.Registry, dlq.DecodeLandRequestID, sc.TopicKey, sc.ConsumerGroup), nil
		},
	},
	{
		Key:           topickey.TopicKeyCancel,
		Name:          "cancel",
		ConsumerGroup: "orchestrator",
		New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
			return cancel.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
		},
		DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
			return dlq.NewDLQRequestController(d.Logger, d.Scope, d.Storage, sc.Registry, dlq.DecodeCancelRequestID, sc.TopicKey, sc.ConsumerGroup), nil
		},
	},
	{
		Key:           topickey.TopicKeyValidate,
		Name:          "validate",
		ConsumerGroup: "orchestrator",
		New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
			return validate.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, d.ChangeProvider, d.Validator, runwaymq.TopicKeyMergeConflictCheck, sc.TopicKey, sc.ConsumerGroup), nil
		},
		DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
			return dlq.NewDLQRequestController(d.Logger, d.Scope, d.Storage, sc.Registry, dlq.DecodeRequestID, sc.TopicKey, sc.ConsumerGroup), nil
		},
	},
	{
		Key:           runwaymq.TopicKeyMergeConflictCheckSignal,
		Name:          "merge-conflict-check-signal",
		ConsumerGroup: "orchestrator",
		New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
			return mergeconflictsignal.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
		},
		DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
			return dlq.NewDLQMergeConflictSignalController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
		},
	},
	{
		Key:           topickey.TopicKeyBatch,
		Name:          "batch",
		ConsumerGroup: "orchestrator",
		New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
			return batch.NewController(d.Logger, d.Scope, sc.Registry, d.Counter, d.Storage, d.Analyzer, sc.TopicKey, sc.ConsumerGroup), nil
		},
		DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
			return dlq.NewDLQRequestController(d.Logger, d.Scope, d.Storage, sc.Registry, dlq.DecodeRequestID, sc.TopicKey, sc.ConsumerGroup), nil
		},
	},
	{
		Key:           topickey.TopicKeySpeculate,
		Name:          "speculate",
		ConsumerGroup: "orchestrator",
		New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
			return speculate.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
		},
		DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
			return dlq.NewDLQBatchController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
		},
	},
	{
		Key:           topickey.TopicKeyBuild,
		Name:          "build",
		ConsumerGroup: "orchestrator",
		New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
			return build.NewController(d.Logger, d.Scope, d.Storage, d.BuildRunner, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
		},
		DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
			return dlq.NewDLQBatchController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
		},
	},
	{
		Key:           topickey.TopicKeyBuildSignal,
		Name:          "buildsignal",
		ConsumerGroup: "orchestrator",
		New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
			return buildsignal.NewController(d.Logger, d.Scope, d.Storage, d.BuildRunner, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
		},
		DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
			return dlq.NewDLQBuildSignalController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
		},
	},
	{
		Key:           topickey.TopicKeyMerge,
		Name:          "submitqueue-merge",
		ConsumerGroup: "orchestrator",
		New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
			return merge.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, runwaymq.TopicKeyMerge, sc.TopicKey, sc.ConsumerGroup), nil
		},
		DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
			return dlq.NewDLQBatchController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
		},
	},
	{
		Key:           runwaymq.TopicKeyMergeSignal,
		Name:          "merge-signal",
		ConsumerGroup: "orchestrator",
		New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
			return mergesignal.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
		},
		DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
			return dlq.NewDLQMergeSignalController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
		},
	},
	{
		Key:           topickey.TopicKeyConclude,
		Name:          "conclude",
		ConsumerGroup: "orchestrator",
		New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
			return conclude.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
		},
		DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
			return dlq.NewDLQBatchController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
		},
	},
}

Stages is the orchestrator's pipeline topology as a typed table. Adding a stage = adding one row. Nothing else, anywhere.

Pipeline: 

start → cancel → validate ⇢ (runway) ⇢ mergeconflictsignal → batch → speculate → build → buildsignal ─┐
                                                                       ↑  ↘             ↻ poll       │
                                                                       │   merge → conclude          │
                                                                       │     │                       │
                                                                       └─────┴───────────────────────┘

Functions

This section is empty.

Types

type Controllers 

type Controllers struct {
	// Ping is the health-check controller.
	Ping *controller.PingController
}

Controllers holds the orchestrator's RPC-facing controllers, constructed but NOT bound to any wire contract. Binding to a proto service + transport is host glue, because deployers may use different protos or transports.

func NewControllers 

func NewControllers(d Deps) Controllers

NewControllers creates the orchestrator's RPC controllers from the given Deps. The PingController takes a base *zap.Logger, so we desugar the SugaredLogger.

type Deps 

type Deps struct {
	// Logger is the structured logger for all controllers.
	Logger *zap.SugaredLogger

	// Scope is the metrics scope for all controllers.
	Scope tally.Scope

	// Storage provides request, batch, and change stores.
	Storage storage.Storage

	// Counter provides distributed batch counters.
	Counter counter.Counter

	// BuildRunner resolves the build runner for each queue.
	BuildRunner buildrunner.Factory

	// ChangeProvider resolves the change provider for each queue.
	ChangeProvider changeprovider.Factory

	// Analyzer resolves the conflict analyzer for each queue.
	Analyzer conflict.Factory

	// Validator resolves the validator for each queue.
	Validator validator.Factory
}

Deps is the full set of dependencies the orchestrator pipeline needs. This struct IS the service's public API toward deployers: fill every field, pass it and Stages to pipeline.Construct, and you get a running orchestrator pipeline.

Source Files

View all Source files

Directories

Path Synopsis
batch
build
buildsignal
Package buildsignal implements the build poll loop.
 Package buildsignal implements the build poll loop.
cancel
Package cancel implements the orchestrator-side cancel controller.
 Package cancel implements the orchestrator-side cancel controller.
conclude
dlq
Package dlq contains controllers that consume messages from per-topic dead-letter queues and reconcile the affected request and batch entities into a terminal failed state.
 Package dlq contains controllers that consume messages from per-topic dead-letter queues and reconcile the affected request and batch entities into a terminal failed state.
merge
Package merge implements the trigger stage for the asynchronous merge.
 Package merge implements the trigger stage for the asynchronous merge.
mergeconflictsignal
Package mergeconflictsignal consumes merge-conflict check results from runway's signal queue, correlates them to the request by the echoed id, and either advances the request to the batch stage (mergeable) or fails it (conflicted).
 Package mergeconflictsignal consumes merge-conflict check results from runway's signal queue, correlates them to the request by the echoed id, and either advances the request to the batch stage (mergeable) or fails it (conflicted).
mergesignal
Package mergesignal consumes merge results from runway's merge-signal queue, correlates them to the batch by the echoed id, and transitions the batch to a terminal state — Succeeded when runway merged the batch, Failed when it could not — then fans the batch out to conclude (so member requests pick up the outcome) and speculate (so dependents can re-plan).
 Package mergesignal consumes merge results from runway's merge-signal queue, correlates them to the batch by the echoed id, and transitions the batch to a terminal state — Succeeded when runway merged the batch, Failed when it could not — then fans the batch out to conclude (so member requests pick up the outcome) and speculate (so dependents can re-plan).
speculate
start
validate

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.