engine

package

v1.16.1-alpha Latest Latest Go to latest Published: Jul 17, 2026 License: MIT Imports: 23 Imported by: 0

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

tangled.org/@tangled.org/core

Links

Open Source Insights

Documentation ¶

Index ¶

Constants
Variables
func DescribeManifestError(raw string, schema any) error
func StartWorkflows(l *slog.Logger, vault secrets.Manager, cfg *config.Config, db *db.DB, ...)
type NoopSlot
- func (NoopSlot) Release()
type ResourceScheduler
- func NewResourceScheduler[R Resources[R]](budget, max R, agingThreshold time.Duration) *ResourceScheduler[R]
- func (s *ResourceScheduler[R]) Acquire(ctx context.Context, req R) (WorkflowSlot, error)
type Resources
type S3
- func NewS3(bucket string) (*S3, error)
- func (s *S3) ReadFile(ctx context.Context, name string) ([]byte, error)
- func (s *S3) WriteFile(ctx context.Context, path string) error
type SemaphoreSlotter
- func NewSemaphoreSlotter(maxConcurrent int) *SemaphoreSlotter
- func (a *SemaphoreSlotter) AcquireWorkflowSlot(ctx context.Context, wid models.WorkflowId, wf *models.Workflow) (WorkflowSlot, error)
type WorkflowSlot
type WorkflowSlotter

Constants ¶

View Source

const BaseS3Path = "spindle/workflows"

Variables ¶

View Source

var (
	ErrTimedOut       = errors.New("timed out")
	ErrWorkflowFailed = errors.New("workflow failed")
)

View Source

var ErrNoWorkflowSlots = errors.New("no workflow slots available")

Functions ¶

func DescribeManifestError ¶

func DescribeManifestError(raw string, schema any) error

this exists because yaml.v3 reports mismatches as "cannot unmarshal !!seq into map[string]interface {}", which is kind of confusing, even if it outputs a line. so we use reflection, walk the node tree alongside the schema type, and point at the field that's actually mis-shaped.

returns nil when nothing is structurally wrong.

func StartWorkflows ¶

func StartWorkflows(l *slog.Logger, vault secrets.Manager, cfg *config.Config, db *db.DB, n *notifier.Notifier, ctx context.Context, pipeline *models.Pipeline, pipelineId models.PipelineId)

Types ¶

type NoopSlot ¶

type NoopSlot struct{}

func (NoopSlot) Release ¶

func (NoopSlot) Release()

type ResourceScheduler ¶

type ResourceScheduler[R Resources[R]] struct {
	// contains filtered or unexported fields
}

func NewResourceScheduler ¶

func NewResourceScheduler[R Resources[R]](budget, max R, agingThreshold time.Duration) *ResourceScheduler[R]

func (*ResourceScheduler[R]) Acquire ¶

func (s *ResourceScheduler[R]) Acquire(ctx context.Context, req R) (WorkflowSlot, error)

type Resources ¶

type Resources[Self any] interface {
	Fits(Self) bool
	Add(Self) Self
	Sub(Self) Self
}

type S3 ¶

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

func NewS3 ¶

func NewS3(bucket string) (*S3, error)

func (*S3) ReadFile ¶

func (s *S3) ReadFile(ctx context.Context, name string) ([]byte, error)

func (*S3) WriteFile ¶

func (s *S3) WriteFile(ctx context.Context, path string) error

type SemaphoreSlotter ¶

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

limit by concurrent workflow count

func NewSemaphoreSlotter ¶

func NewSemaphoreSlotter(maxConcurrent int) *SemaphoreSlotter

func (*SemaphoreSlotter) AcquireWorkflowSlot ¶

func (a *SemaphoreSlotter) AcquireWorkflowSlot(ctx context.Context, wid models.WorkflowId, wf *models.Workflow) (WorkflowSlot, error)

type WorkflowSlot ¶

type WorkflowSlot interface {
	Release()
}

type WorkflowSlotter ¶

type WorkflowSlotter interface {
	AcquireWorkflowSlot(ctx context.Context, wid models.WorkflowId, wf *models.Workflow) (WorkflowSlot, error)
}

Source Files ¶

engine.go
manifest.go
s3.go
scheduler.go
slot.go

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