store

package
v0.22.0 Latest Latest
Warning

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

 Go to latest Published: Nov 6, 2026 License: Apache-2.0 Imports: 19 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	// SyncStatusSynced indicates that a container record is synchronised with the Docker daemon. The record may
	// become outdated even when the status is "synced" if the machine crashes or a network partition occurs.
	// The cluster membership state of the machine should also be checked to determine if the record can be trusted.
	SyncStatusSynced = "synced"
	// SyncStatusOutdated indicates that a container record may be outdated, for example, due to being unable
	// to retrieve the container's state from the Docker daemon or when the machine is being stopped or restarted.
	SyncStatusOutdated = "outdated"
)

Variables

View Source 
var (
	//go:embed schema.sql
	Schema string

	ErrKeyNotFound     = errors.New("key not found")
	ErrMachineNotFound = errors.New("machine not found")
)
View Source 
var ErrInvalidStoreVersion = errors.New("invalid store version")

ErrInvalidStoreVersion indicates an invalid actor UUID in a store version vector.

Functions

This section is empty.

Types

type ContainerRecord 

type ContainerRecord struct {
	Container  api.ServiceContainer
	MachineID  string
	SyncStatus string
	UpdatedAt  time.Time
}

type DeleteOptions 

type DeleteOptions struct {
	// IDs filters containers by their container IDs.
	IDs []string
	// MachineIDs filters containers by the machine IDs they are running on.
	MachineIDs []string
}

type Keyspace added in v0.21.0 

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

Keyspace provides namespace-scoped key-value storage. Keys are relative to the namespace. Prefix operations use simple string-prefix matching.

func (*Keyspace) Delete added in v0.21.0 

func (k *Keyspace) Delete(ctx context.Context, key string, opts KeyspaceDeleteOptions) error

Delete deletes key. When Prefix is true, it also deletes every key that starts with key.

func (*Keyspace) Get added in v0.21.0 

func (k *Keyspace) Get(ctx context.Context, key string) (Record, error)

Get returns the entry for key. It returns ErrKeyNotFound when the key does not exist.

func (*Keyspace) List added in v0.21.0 

func (k *Keyspace) List(ctx context.Context, prefix string, opts KeyspaceListOptions) ([]Record, error)

List returns entries whose keys start with prefix, ordered by key. An empty prefix lists every entry in the keyspace.

func (*Keyspace) Put added in v0.21.0 

func (k *Keyspace) Put(ctx context.Context, key string, value []byte) error

Put stores value at key.

type KeyspaceDeleteOptions added in v0.21.0 

type KeyspaceDeleteOptions struct {
	// Prefix deletes the key and every key that starts with it.
	Prefix bool
}

KeyspaceDeleteOptions controls which records Keyspace.Delete deletes.

type KeyspaceListOptions added in v0.21.0 

type KeyspaceListOptions struct {
	// KeysOnly omits record values from the query when true.
	KeysOnly bool
}

KeyspaceListOptions controls the records returned by Keyspace.List.

type ListOptions 

type ListOptions struct {
	// MachineIDs filters containers by the machine IDs they are running on.
	MachineIDs      []string
	ServiceIDOrName ServiceIDOrNameOptions
}

type Record added in v0.21.0 

type Record struct {
	Key       string
	Value     []byte
	UpdatedAt time.Time
}

Record is a key-value entry and its metadata.

type ServiceIDOrNameOptions 

type ServiceIDOrNameOptions struct {
	ID   string
	Name string
}

ServiceIDOrNameOptions filters containers by the service ID or name they are part of. If both ID and Name are provided, they are combined with an OR operator.

type Store 

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

Store is a cluster store backed by a distributed Corrosion database.

func New 

func New(corro *corrosion.APIClient, corroAdmin *corrosion.AdminClient) *Store

func (*Store) CreateMachine 

func (s *Store) CreateMachine(ctx context.Context, m *pb.MachineInfo) error

func (*Store) CreateOrUpdateContainer 

func (s *Store) CreateOrUpdateContainer(ctx context.Context, ctr api.ServiceContainer, machineID string) error

CreateOrUpdateContainer creates a new container record or updates an existing one in the store database. The container is associated with the given machine ID that indicates which machine the container is running on.

func (*Store) Delete deprecated 

func (s *Store) Delete(ctx context.Context, key string) error

Delete deletes an unnamespaced legacy value.

Deprecated: Existing callers may continue to use Delete for legacy records. New code should use Store.Keyspace.

func (*Store) DeleteContainers 

func (s *Store) DeleteContainers(ctx context.Context, opts DeleteOptions) error

DeleteContainers deletes container records from the store database that match the given options. If no filter is set, all container records are deleted. Filters are combined with AND.

func (*Store) DeleteMachine added in v0.10.0 

func (s *Store) DeleteMachine(ctx context.Context, id string) error

func (*Store) Get deprecated 

func (s *Store) Get(ctx context.Context, key string, value any) error

Get retrieves an unnamespaced legacy value.

Deprecated: Existing callers may continue to use Get for legacy records. New code should use Store.Keyspace.

func (*Store) GetMachine added in v0.10.0 

func (s *Store) GetMachine(ctx context.Context, machineID string) (*pb.MachineInfo, error)

func (*Store) Keyspace added in v0.21.0 

func (s *Store) Keyspace(namespace string) (*Keyspace, error)

Keyspace returns key-value storage scoped to namespace.

func (*Store) ListContainers 

func (s *Store) ListContainers(ctx context.Context, opts ListOptions) ([]ContainerRecord, error)

ListContainers returns a list of container records from the store database that match the given options. The result excludes orphan containers whose machine is no longer in the cluster.

func (*Store) ListMachines 

func (s *Store) ListMachines(ctx context.Context) ([]*pb.MachineInfo, error)

func (*Store) Put deprecated 

func (s *Store) Put(ctx context.Context, key string, value any) error

Put stores an unnamespaced legacy value.

Deprecated: Existing callers may continue to use Put for legacy records. New code should use Store.Keyspace.

func (*Store) SubscribeContainers 

func (s *Store) SubscribeContainers(ctx context.Context) ([]ContainerRecord, <-chan struct{}, error)

SubscribeContainers returns a list of containers and a channel that signals changes to the list. The channel doesn't receive any values, it just signals when a container(s) has been added, updated, or deleted in the database. The result excludes orphan containers whose machine is no longer in the cluster. The channel is closed when the containers are no longer subscribable: either the provided context is cancelled or the underlying subscription fails.

func (*Store) SubscribeMachines 

func (s *Store) SubscribeMachines(ctx context.Context) ([]*pb.MachineInfo, <-chan struct{}, error)

SubscribeMachines returns a list of machines and a channel that signals changes to the list. The channel doesn't receive any values, it just signals when a machine has been added, updated, or deleted in the database. The channel is closed when the machines are no longer subscribable: either the provided context is cancelled or the underlying subscription fails.

func (*Store) UpdateMachine added in v0.10.0 

func (s *Store) UpdateMachine(ctx context.Context, m *pb.MachineInfo) error

func (*Store) Version added in v0.20.0 

func (s *Store) Version(ctx context.Context) (api.StoreVersion, error)

Version returns the replication progress observed by this store. Each entry maps a Corrosion actor UUID to its highest processed database version. Corrosion also counts versions whose changes were superseded and skipped. Versions at or below a reported version may still be missing or pending locally.

You can use Store.WaitForVersion to wait for replication through these versions, subject to the data-availability limitations documented there.

Capturing a vector does not wait for replication or prevent further writes.

func (*Store) WaitForVersion added in v0.21.0 

func (s *Store) WaitForVersion(ctx context.Context, minVersion api.StoreVersion) error

WaitForVersion waits until the local store has reached each actor's minimum version in minVersion, with no known missing or pending transactions through those versions from active members. Corrosion may satisfy a version by applying its surviving changes or by marking it complete because its changes have been superseded.

Waiting normally makes the captured data available locally. However, if another write replaces some of that data before it arrives, Corrosion can complete the older version without transferring the replaced data. If the replacement is outside minVersion, this method can return while the affected data is still missing or outdated. This can happen during concurrent updates even when all machines are well connected.

This method does not guarantee an exact snapshot or delivery of every historical value. Callers that require a specific record or condition should verify it after waiting.

The method observes native replication without initiating synchronisation. An empty minVersion requires no replication. The context controls cancellation and the deadline.

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.