forge_lib_kvtx

package
v0.53.0 Latest Latest
Warning

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

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

Documentation

Index

Constants

View Source 
const ConfigID = ControllerID

ConfigID is the string used to identify this config object.

View Source 
const ControllerID = "forge/lib/kvtx"

ControllerID is the ID of the controller.

Variables

View Source 
var (
	OpType_name = map[int32]string{
		0: "OpType_NONE",
		1: "OpType_GET",
		2: "OpType_GET_EXISTS",
		3: "OpType_CHECK",
		4: "OpType_CHECK_BLOB",
		5: "OpType_CHECK_EXISTS",
		6: "OpType_CHECK_NOT_EXISTS",
		7: "OpType_SET",
		8: "OpType_SET_BLOB",
		9: "OpType_DELETE",
	}
	OpType_value = map[string]int32{
		"OpType_NONE":             0,
		"OpType_GET":              1,
		"OpType_GET_EXISTS":       2,
		"OpType_CHECK":            3,
		"OpType_CHECK_BLOB":       4,
		"OpType_CHECK_EXISTS":     5,
		"OpType_CHECK_NOT_EXISTS": 6,
		"OpType_SET":              7,
		"OpType_SET_BLOB":         8,
		"OpType_DELETE":           9,
	}
)

Enum value maps for OpType.

View Source 
var (
	// ErrEmptyKey is returned if the key was not specified in the op
	ErrEmptyKey = errors.New("key was not specified for operation")
	// ErrUnknownOpType is returned if the operation type was unknown.
	ErrUnknownOpType = errors.New("operation type was unknown")
	// ErrValueMismatch is returned if the values did not match.
	ErrValueMismatch = errors.New("values did not match")
)
View Source 
var Version = controller.MustParseVersion("0.0.1")

Version is the version of the controller implementation.

Functions

func AddFactories 

func AddFactories(b bus.Bus, sr *static.Resolver)

AddFactories adds factories to an existing static resolver.

func ApplyOpCheck 

func ApplyOpCheck(
	ctx context.Context,
	handle forge_target.ExecControllerHandle,
	btx kvtx.BlockTx,
	key []byte,
	value *forge_value.Value,
	valueIsBlob bool,
	outputName string,
) error

ApplyOpCheck applies a CHECK operation against a store.

func ApplyOpCheckExists 

func ApplyOpCheckExists(
	ctx context.Context,
	handle forge_target.ExecControllerHandle,
	btx kvtx.BlockTx,
	key []byte,
	shouldExist bool,
) error

ApplyOpCheckExists ensures a key does or does not exist.

func ApplyOpDelete 

func ApplyOpDelete(
	ctx context.Context,
	handle forge_target.ExecControllerHandle,
	btx kvtx.BlockTx,
	key []byte,
	outputName string,
) error

ApplyOpDelete applies a DELETE operation against a store.

func ApplyOpGet 

func ApplyOpGet(
	ctx context.Context,
	handle forge_target.ExecControllerHandle,
	btx kvtx.BlockTx,
	key []byte,
	outputName string,
) error

ApplyOpGet applies a GET operation against a store.

func ApplyOpGetExists 

func ApplyOpGetExists(
	ctx context.Context,
	handle forge_target.ExecControllerHandle,
	btx kvtx.BlockTx,
	key []byte,
	outputName string,
) error

ApplyOpGetExists applies a EXISTS operation against a store.

func ApplyOpSet 

func ApplyOpSet(
	ctx context.Context,
	handle forge_target.ExecControllerHandle,
	btx kvtx.BlockTx,
	key []byte,
	value *forge_value.Value,
	valueIsBlob bool,
	outputName string,
) error

ApplyOpSet applies a SET operation against a store. sets a reference to the location.

func NewConfigInputBlock 

func NewConfigInputBlock() block.Block

NewConfigInputBlock constructs a new ConfigInput block.

Types

type Config 

type Config struct {

	// Ops is the list of operations to apply.
	Ops []*Op `protobuf:"bytes,1,rep,name=ops,proto3" json:"ops,omitempty"`
	// ConfigInput is the name of an Input to load additional ops from.
	// The referenced block should contain a ConfigInput object.
	ConfigInput string `protobuf:"bytes,2,opt,name=config_input,json=configInput,proto3" json:"configInput,omitempty"`
	// IgnoreErrors warns on errors and continues w/o failing.
	IgnoreErrors bool `protobuf:"varint,3,opt,name=ignore_errors,json=ignoreErrors,proto3" json:"ignoreErrors,omitempty"`
	// contains filtered or unexported fields
}

Config is the configuration for the Forge KVTX exec controller. Implements get, set, delete, keys against a k/v block graph. Operations are sequential, failed ops cancel the entire sequence.

Inputs:

  • store: reference to a kvtx_block.KeyValueStore.
  • ... config_input as configured below (optional).
  • ... any operation inputs configured below

Outputs:

  • store: output (modified) KeyValueStore.
  • ... any operation outputs configured below.

func (*Config) CloneMessageVT 

func (m *Config) CloneMessageVT() protobuf_go_lite.CloneMessage

func (*Config) CloneVT 

func (m *Config) CloneVT() *Config

func (*Config) EqualMessageVT 

func (this *Config) EqualMessageVT(thatMsg any) bool

func (*Config) EqualVT 

func (this *Config) EqualVT(that *Config) bool

func (*Config) EqualsConfig 

func (c *Config) EqualsConfig(other config.Config) bool

EqualsConfig checks if the other config is equal.

func (*Config) GetConfigID 

func (c *Config) GetConfigID() string

GetConfigID returns the unique string for this configuration type. This string is stored with the encoded config.

func (*Config) GetConfigInput 

func (x *Config) GetConfigInput() string

func (*Config) GetIgnoreErrors 

func (x *Config) GetIgnoreErrors() bool

func (*Config) GetOps 

func (x *Config) GetOps() []*Op

func (*Config) IsEmpty 

func (c *Config) IsEmpty() bool

IsEmpty checks if there are no operations in the config.

func (*Config) MarshalBlock 

func (c *Config) MarshalBlock() ([]byte, error)

MarshalBlock marshals the block to binary. This is the initial step of marshaling, before transformations.

func (*Config) MarshalJSON 

func (x *Config) MarshalJSON() ([]byte, error)

MarshalJSON marshals the Config to JSON.

func (*Config) MarshalProtoJSON 

func (x *Config) MarshalProtoJSON(s *json.MarshalState)

MarshalProtoJSON marshals the Config message to JSON.

func (*Config) MarshalProtoText 

func (x *Config) MarshalProtoText() string

func (*Config) MarshalToSizedBufferVT 

func (m *Config) MarshalToSizedBufferVT(dAtA []byte) (int, error)

func (*Config) MarshalToVT 

func (m *Config) MarshalToVT(dAtA []byte) (int, error)

func (*Config) MarshalVT 

func (m *Config) MarshalVT() (dAtA []byte, err error)

func (*Config) ProtoMessage 

func (*Config) ProtoMessage()

func (*Config) Reset 

func (x *Config) Reset()

func (*Config) SizeVT 

func (m *Config) SizeVT() (n int)

func (*Config) String 

func (x *Config) String() string

func (*Config) UnmarshalBlock 

func (c *Config) UnmarshalBlock(data []byte) error

UnmarshalBlock unmarshals the block to the object. This is the final step of decoding, after transformations.

func (*Config) UnmarshalJSON 

func (x *Config) UnmarshalJSON(b []byte) error

UnmarshalJSON unmarshals the Config from JSON.

func (*Config) UnmarshalProtoJSON 

func (x *Config) UnmarshalProtoJSON(s *json.UnmarshalState)

UnmarshalProtoJSON unmarshals the Config message from JSON.

func (*Config) UnmarshalVT 

func (m *Config) UnmarshalVT(dAtA []byte) error

func (*Config) Validate 

func (c *Config) Validate() error

Validate validates the configuration.

type ConfigInput 

type ConfigInput struct {

	// Ops is the list of operations to apply.
	Ops []*Op `protobuf:"bytes,1,rep,name=ops,proto3" json:"ops,omitempty"`
	// contains filtered or unexported fields
}

ConfigInput is a block containing additional ops from an Input.

func FetchConfigInput 

func FetchConfigInput(
	ctx context.Context,
	handle forge_target.ExecControllerHandle,
	val *forge_value.Value,
) (*ConfigInput, error)

FetchConfigInput fetches the ConfigInput block from the Value.

func UnmarshalConfigInput 

func UnmarshalConfigInput(ctx context.Context, bcs *block.Cursor) (*ConfigInput, error)

UnmarshalConfigInput unmarshals a config input block from the cursor.

func (*ConfigInput) CloneMessageVT 

func (m *ConfigInput) CloneMessageVT() protobuf_go_lite.CloneMessage

func (*ConfigInput) CloneVT 

func (m *ConfigInput) CloneVT() *ConfigInput

func (*ConfigInput) EqualMessageVT 

func (this *ConfigInput) EqualMessageVT(thatMsg any) bool

func (*ConfigInput) EqualVT 

func (this *ConfigInput) EqualVT(that *ConfigInput) bool

func (*ConfigInput) GetOps 

func (x *ConfigInput) GetOps() []*Op

func (*ConfigInput) MarshalBlock 

func (b *ConfigInput) MarshalBlock() ([]byte, error)

MarshalBlock marshals the block to binary. This is the initial step of marshaling, before transformations.

func (*ConfigInput) MarshalJSON 

func (x *ConfigInput) MarshalJSON() ([]byte, error)

MarshalJSON marshals the ConfigInput to JSON.

func (*ConfigInput) MarshalProtoJSON 

func (x *ConfigInput) MarshalProtoJSON(s *json.MarshalState)

MarshalProtoJSON marshals the ConfigInput message to JSON.

func (*ConfigInput) MarshalProtoText 

func (x *ConfigInput) MarshalProtoText() string

func (*ConfigInput) MarshalToSizedBufferVT 

func (m *ConfigInput) MarshalToSizedBufferVT(dAtA []byte) (int, error)

func (*ConfigInput) MarshalToVT 

func (m *ConfigInput) MarshalToVT(dAtA []byte) (int, error)

func (*ConfigInput) MarshalVT 

func (m *ConfigInput) MarshalVT() (dAtA []byte, err error)

func (*ConfigInput) ProtoMessage 

func (*ConfigInput) ProtoMessage()

func (*ConfigInput) Reset 

func (x *ConfigInput) Reset()

func (*ConfigInput) SizeVT 

func (m *ConfigInput) SizeVT() (n int)

func (*ConfigInput) String 

func (x *ConfigInput) String() string

func (*ConfigInput) UnmarshalBlock 

func (b *ConfigInput) UnmarshalBlock(data []byte) error

UnmarshalBlock unmarshals the block to the object. This is the final step of decoding, after transformations.

func (*ConfigInput) UnmarshalJSON 

func (x *ConfigInput) UnmarshalJSON(b []byte) error

UnmarshalJSON unmarshals the ConfigInput from JSON.

func (*ConfigInput) UnmarshalProtoJSON 

func (x *ConfigInput) UnmarshalProtoJSON(s *json.UnmarshalState)

UnmarshalProtoJSON unmarshals the ConfigInput message from JSON.

func (*ConfigInput) UnmarshalVT 

func (m *ConfigInput) UnmarshalVT(dAtA []byte) error

type Controller 

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

Controller implements the kvtx operations controller.

func NewController 

func NewController(
	le *logrus.Entry,
	bus bus.Bus,
	conf *Config,
) *Controller

NewController constructs a new kvtx ops execution controller.

func (*Controller) Close 

func (c *Controller) Close() error

Close releases any resources used by the controller. Error indicates any issue encountered releasing.

func (*Controller) Execute 

func (c *Controller) Execute(ctx context.Context) error

Execute executes the controller goroutine. Returning nil ends execution. Returning an error triggers a retry with backoff.

func (*Controller) GetControllerInfo 

func (c *Controller) GetControllerInfo() *controller.Info

GetControllerInfo returns information about the controller.

func (*Controller) HandleDirective 

func (c *Controller) HandleDirective(ctx context.Context, inst directive.Instance) ([]directive.Resolver, error)

HandleDirective asks if the handler can resolve the directive.

func (*Controller) InitForgeExecController 

func (c *Controller) InitForgeExecController(
	ctx context.Context,
	inputVals forge_target.InputMap,
	handle forge_target.ExecControllerHandle,
) error

InitForgeExecController initializes the Forge execution controller. This is called before Execute(). Any error returned cancels execution of the controller.

type Factory 

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

Factory constructs a kvtx controller.

func NewFactory 

func NewFactory(bus bus.Bus) *Factory

NewFactory builds a factory.

func (*Factory) Construct 

func (t *Factory) Construct(
	ctx context.Context,
	conf config.Config,
	opts controller.ConstructOpts,
) (controller.Controller, error)

Construct constructs the associated controller given configuration.

func (*Factory) ConstructConfig 

func (t *Factory) ConstructConfig() config.Config

ConstructConfig constructs an instance of the controller configuration.

func (*Factory) GetConfigID 

func (t *Factory) GetConfigID() string

GetConfigID returns the configuration ID for the controller.

func (*Factory) GetControllerID 

func (t *Factory) GetControllerID() string

GetControllerID returns the unique ID for the controller.

func (*Factory) GetVersion 

func (t *Factory) GetVersion() controller.Version

GetVersion returns the version of this controller.

type Op 

type Op struct {

	// OpType is the operation type to apply.
	OpType OpType `protobuf:"varint,1,opt,name=op_type,json=opType,proto3" json:"opType,omitempty"`
	// KeyInput is the name of the Input to use for the Key.
	// The raw block data will be used as the key.
	KeyInput string `protobuf:"bytes,2,opt,name=key_input,json=keyInput,proto3" json:"keyInput,omitempty"`
	// Key is the in-line configured key to use.
	// Converted to []byte without nil terminator.
	// If set, overrides input_key.
	Key string `protobuf:"bytes,3,opt,name=key,proto3" json:"key,omitempty"`
	// ValueInput is the name of the Input to use for the Value argument.
	//
	// - CHECK: contains the value to check against the GET value.
	// - SET: contains the value to store.
	//
	// If SET, stores the block reference into the tree.
	// If SET_BLOB, stores the block reference into the tree, marking it as a Blob
	// If CHECK, compares the given block reference with the stored.
	// Note: overridden by inline value if set and not empty.
	// Note: overridden by value_string if set and not empty.
	ValueInput string `protobuf:"bytes,6,opt,name=value_input,json=valueInput,proto3" json:"valueInput,omitempty"`
	// Value is the in-line value argument.
	//
	// See input field for additional notes.
	Value *value.Value `protobuf:"bytes,7,opt,name=value,proto3" json:"value,omitempty"`
	// ValueString is an in-line string specification of the value.
	// Note: overridden by inline value if set and not empty.
	ValueString string `protobuf:"bytes,8,opt,name=value_string,json=valueString,proto3" json:"valueString,omitempty"`
	// Output is the name of the output to use for the Value.
	//
	// - GET: contains the value retrieved from the store.
	// - CHECK: same as GET, contains the retrieved value.
	// - SET: contains the value added to the store.
	// - DELETE: contains the value just before deleting.
	//
	// Note: contains the single most recently set value only.
	// Note: stores the block reference as the output Value.
	Output string `protobuf:"bytes,9,opt,name=output,proto3" json:"output,omitempty"`
	// Ops is the list of sub-operations to apply.
	// The sub-operations inherit the parent config.
	Ops []*Op `protobuf:"bytes,10,rep,name=ops,proto3" json:"ops,omitempty"`
	// contains filtered or unexported fields
}

Op is an operation definition.

func (*Op) CloneMessageVT 

func (m *Op) CloneMessageVT() protobuf_go_lite.CloneMessage

func (*Op) CloneVT 

func (m *Op) CloneVT() *Op

func (*Op) EqualMessageVT 

func (this *Op) EqualMessageVT(thatMsg any) bool

func (*Op) EqualVT 

func (this *Op) EqualVT(that *Op) bool

func (*Op) GetKey 

func (x *Op) GetKey() string

func (*Op) GetKeyInput 

func (x *Op) GetKeyInput() string

func (*Op) GetOpType 

func (x *Op) GetOpType() OpType

func (*Op) GetOps 

func (x *Op) GetOps() []*Op

func (*Op) GetOutput 

func (x *Op) GetOutput() string

func (*Op) GetValue 

func (x *Op) GetValue() *value.Value

func (*Op) GetValueInput 

func (x *Op) GetValueInput() string

func (*Op) GetValueString 

func (x *Op) GetValueString() string

func (*Op) IsEmpty 

func (o *Op) IsEmpty() bool

IsEmpty checks if the configuration is empty.

func (*Op) MarshalJSON 

func (x *Op) MarshalJSON() ([]byte, error)

MarshalJSON marshals the Op to JSON.

func (*Op) MarshalProtoJSON 

func (x *Op) MarshalProtoJSON(s *json.MarshalState)

MarshalProtoJSON marshals the Op message to JSON.

func (*Op) MarshalProtoText 

func (x *Op) MarshalProtoText() string

func (*Op) MarshalToSizedBufferVT 

func (m *Op) MarshalToSizedBufferVT(dAtA []byte) (int, error)

func (*Op) MarshalToVT 

func (m *Op) MarshalToVT(dAtA []byte) (int, error)

func (*Op) MarshalVT 

func (m *Op) MarshalVT() (dAtA []byte, err error)

func (*Op) ProtoMessage 

func (*Op) ProtoMessage()

func (*Op) Reset 

func (x *Op) Reset()

func (*Op) SizeVT 

func (m *Op) SizeVT() (n int)

func (*Op) String 

func (x *Op) String() string

func (*Op) UnmarshalJSON 

func (x *Op) UnmarshalJSON(b []byte) error

UnmarshalJSON unmarshals the Op from JSON.

func (*Op) UnmarshalProtoJSON 

func (x *Op) UnmarshalProtoJSON(s *json.UnmarshalState)

UnmarshalProtoJSON unmarshals the Op message from JSON.

func (*Op) UnmarshalVT 

func (m *Op) UnmarshalVT(dAtA []byte) error

func (*Op) Validate 

func (o *Op) Validate() error

Validate validates the configuration. Note: recursively checks nested ops.

type OpQueue 

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

OpQueue prefetches input values for a set of ops.

func NewOpQueue 

func NewOpQueue(
	ctx context.Context,
	inputVals forge_target.InputMap,
	handle forge_target.ExecControllerHandle,
) *OpQueue

NewOpQueue constructs a new op queue.

func (*OpQueue) AddOp 

func (q *OpQueue) AddOp(op *Op) error

AddOp applies an operation to the queue.

func (*OpQueue) AddOps 

func (q *OpQueue) AddOps(ops []*Op) error

AddOps adds a list of ops to the queue. Resolves the input values or returns an error.

func (*OpQueue) AddPendingOp 

func (q *OpQueue) AddPendingOp(op *PendingOp) int

AddPendingOp queues a pending op.

func (*OpQueue) ApplyOps 

func (q *OpQueue) ApplyOps(btx kvtx.BlockTx, clear, ignoreErr bool) error

ApplyOps applies all pending ops.

func (*OpQueue) GetPendingOps 

func (q *OpQueue) GetPendingOps() []*PendingOp

GetPendingOps returns the slice of pending ops.

type OpType 

type OpType int32

OpType is the list of operation codes for kvtx. 

const (
	// OpType_NONE indicates this Op just sets values for nested ops.
	OpType_OpType_NONE OpType = 0
	// OpType_GET retrieves a BlockRef from the store and copies to output.
	OpType_OpType_GET OpType = 1
	// OpType_GET_EXISTS checks if the key exists and writes a msgpack boolean block to output.
	OpType_OpType_GET_EXISTS OpType = 2
	// OpType_CHECK performs a GET operation and checks the value against the input.
	// The input must be a non-empty Cursor.
	// Compares the block reference stored with the given block ref.
	OpType_OpType_CHECK OpType = 3
	// OpType_CHECK_BLOB performs a GET operation and checks the raw value against the input.
	// The input must be a non-empty Blob.
	// Compares the blob contents with the input blob contents.
	OpType_OpType_CHECK_BLOB OpType = 4
	// OpType_CHECK_EXISTS checks if the given key exists.
	OpType_OpType_CHECK_EXISTS OpType = 5
	// OpType_CHECK_NOT_EXISTS ensures the given key does not exist.
	OpType_OpType_CHECK_NOT_EXISTS OpType = 6
	// OpType_SET sets a BlockRef to the block referenced by the Value.
	// Note: does not handle Blob values, sets as a reference only.
	OpType_OpType_SET OpType = 7
	// OpType_SET_BLOB treats the input BlockRef as a Blob and stores it.
	// The GET operation will return the Blob contents.
	OpType_OpType_SET_BLOB OpType = 8
	// OpType_DELETE deletes a key from the store.
	// The old value is written to the output, if set.
	OpType_OpType_DELETE OpType = 9
)

func (OpType) Enum 

func (x OpType) Enum() *OpType

func (OpType) MarshalJSON 

func (x OpType) MarshalJSON() ([]byte, error)

MarshalJSON marshals the OpType to JSON.

func (OpType) MarshalProtoJSON 

func (x OpType) MarshalProtoJSON(s *json.MarshalState)

MarshalProtoJSON marshals the OpType to JSON.

func (OpType) MarshalProtoText 

func (x OpType) MarshalProtoText() string

func (OpType) MarshalText 

func (x OpType) MarshalText() ([]byte, error)

MarshalText marshals the OpType to text.

func (OpType) String 

func (x OpType) String() string

func (*OpType) UnmarshalJSON 

func (x *OpType) UnmarshalJSON(b []byte) error

UnmarshalJSON unmarshals the OpType from JSON.

func (*OpType) UnmarshalProtoJSON 

func (x *OpType) UnmarshalProtoJSON(s *json.UnmarshalState)

UnmarshalProtoJSON unmarshals the OpType from JSON.

func (*OpType) UnmarshalText 

func (x *OpType) UnmarshalText(b []byte) error

UnmarshalText unmarshals the OpType from text.

type PendingOp 

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

PendingOp contains a pending operation with arguments.

func NewPendingOp 

func NewPendingOp(
	opType OpType,
	key []byte,
	value *forge_value.Value,
	valueIsBlob bool,
	outputName string,
) *PendingOp

NewPendingOp constructs a new pending op. value or outputName can be empty depending on the operation.

func (*PendingOp) Apply 

func (o *PendingOp) Apply(
	ctx context.Context,
	handle forge_target.ExecControllerHandle,
	btx kvtx.BlockTx,
) error

Apply applies the operation to the store.

Source Files

View all Source files

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.