Documentation
¶
Overview ¶
Package dagstate defines persistent state shared across DAG runs.
Index ¶
Constants ¶
This section is empty.
Variables ¶
Functions ¶
func HashValue ¶
func HashValue(value json.RawMessage) string
HashValue returns the SHA-256 hash of a normalized state value.
func NormalizeValue ¶
func NormalizeValue(data []byte) (json.RawMessage, error)
NormalizeValue validates and compacts a JSON value before storage.
func ValidateKeyPrefix ¶
ValidateKeyPrefix rejects malformed key prefixes for list filters.
Types ¶
type Entry ¶
type Entry struct {
Ref
Value json.RawMessage `json:"value"`
Version int64 `json:"version"`
Hash string `json:"hash"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
UpdatedBy *UpdateSource `json:"updated_by,omitempty"`
}
Entry is a versioned JSON value stored for a state reference.
type ListOptions ¶
ListOptions filters state entries by scope, namespace, and key prefix.
func (ListOptions) RecordIDPrefix ¶
func (o ListOptions) RecordIDPrefix() (string, error)
RecordIDPrefix returns the storage prefix for a list filter.
func (ListOptions) Validate ¶
func (o ListOptions) Validate() error
Validate rejects malformed list filters before store access.
type PutOptions ¶
type PutOptions struct {
ExpectedVersion *int64
CreateOnly bool
UpdatedBy *UpdateSource
}
PutOptions controls optimistic concurrency and audit metadata for writes.
type Ref ¶
type Ref struct {
Scope Scope `json:"scope"`
Namespace string `json:"namespace"`
Key string `json:"key"`
}
Ref identifies one persistent DAG state entry.
func RefFromRecordID ¶
RefFromRecordID parses a storage key back into a state reference.
type Scope ¶
type Scope string
Scope identifies the namespace strategy for a state entry.
const ( // ScopeDAG stores state under the current DAG name. ScopeDAG Scope = "dag" // ScopeRootDAG stores state under the root DAG name for nested runs. ScopeRootDAG Scope = "root_dag" // ScopeGlobal stores state in a process-wide namespace. ScopeGlobal Scope = "global" // ScopeCustom stores state in an explicitly provided namespace. ScopeCustom Scope = "custom" // DefaultGlobalNamespace is used when global state does not need a user namespace. DefaultGlobalNamespace = "_" // DefaultMaxValueBytes is the maximum normalized JSON payload size for one state entry. DefaultMaxValueBytes = 1 << 20 )
type Store ¶
type Store interface {
Get(ctx context.Context, ref Ref) (*Entry, error)
Put(ctx context.Context, ref Ref, value json.RawMessage, opts PutOptions) (*Entry, error)
Delete(ctx context.Context, ref Ref) (bool, error)
List(ctx context.Context, opts ListOptions) ([]*Entry, error)
}
Store persists JSON state entries across DAG runs.
type UpdateSource ¶
type UpdateSource struct {
DAGName string `json:"dag_name,omitempty"`
DAGRunID string `json:"dag_run_id,omitempty"`
AttemptID string `json:"attempt_id,omitempty"`
StepName string `json:"step_name,omitempty"`
}
UpdateSource records the DAG run and step that last updated an entry.
func (*UpdateSource) Clone ¶
func (u *UpdateSource) Clone() *UpdateSource
Clone returns a copy of the update source.
Source Files