Help us shape the future of Go! Take the 2026 Go Developer Survey

omashu

package module
v0.2.0 Latest Latest
Warning

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

 Go to latest Published: May 17, 2026 License: Apache-2.0 Imports: 32 Imported by: 0

README

omashu

Documentation

Index

Constants

View Source 
const (
	DefaultProposeTimeout = 5 * time.Second

	// Subdirectories for different components of the store
	DBSubDir   = "db"
	WALSubDir  = "wal"
	SnapSubDir = "snap"
)
View Source 
const (
	RetryCount        = 5
	ReadStatesTimeout = time.Second
)
View Source 
const (
	LOGICAL_BITS = 18
	MAX_LOGICAL  = (1 << LOGICAL_BITS) - 1
)
View Source 
const (
	MaxBatchSize = 1000
)

Variables

View Source 
var (
	// Config validation errors
	ErrMissingRaftConf     = errors.New("raft config required for distributed db")
	ErrMissingCluster      = errors.New("cluster details are missing")
	ErrMissingBaseDir      = errors.New("baseDir is missing")
	ErrMissingSchemaConfig = errors.New("schemaConfig is missing")

	ErrNotLeader      = errors.New("operation can only be performed on leader node")
	ErrProposeTimeout = errors.New("raft propose timeout")

	// ErrBatchTooBig indicates that the batch size exceeds the maximum limit.
	ErrBatchTooBig = errors.New("batch size exceeds maximum limit")

	// ErrUnknownOp indicates that an unknown operation was encountered.
	ErrUnknownOp = errors.New("unknown operation")

	ErrUnknownProtoMsg = errors.New("unknown protobuf message type")
)

Functions

This section is empty.

Types

type Badger 

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

func NewBadger 

func NewBadger(ctx context.Context, cfg *Config) (*Badger, error)

func (*Badger) BulkGet 

func (bdb *Badger) BulkGet(ctx context.Context, keys []string) (map[string][]byte, error)

func (*Badger) Close 

func (bdb *Badger) Close(ctx context.Context)

func (*Badger) Count 

func (bdb *Badger) Count(ctx context.Context, prefix string) int

func (*Badger) DecrBy 

func (bdb *Badger) DecrBy(ctx context.Context, key string, delta uint64) error

func (*Badger) DecrByWithTxn 

func (bdb *Badger) DecrByWithTxn(ctx context.Context, txn *badger.Txn, key string, delta uint64) error

func (*Badger) Delete 

func (bdb *Badger) Delete(ctx context.Context, key string) error

func (*Badger) DeleteByPrefix 

func (bdb *Badger) DeleteByPrefix(ctx context.Context, prefix string) error

func (*Badger) DeleteByPrefixWithTxn 

func (bdb *Badger) DeleteByPrefixWithTxn(ctx context.Context, txn *badger.Txn, prefix string) error

func (*Badger) DeleteWithTxn 

func (bdb *Badger) DeleteWithTxn(ctx context.Context, txn *badger.Txn, key string) error

func (*Badger) Exists 

func (bdb *Badger) Exists(ctx context.Context, key string) bool

func (*Badger) Get 

func (bdb *Badger) Get(ctx context.Context, key string) ([]byte, bool, error)

func (*Badger) GetBadger 

func (bdb *Badger) GetBadger() *badger.DB

func (*Badger) GetByPrefix 

func (bdb *Badger) GetByPrefix(ctx context.Context, prefix string) (map[string][]byte, error)

func (*Badger) GetByPrefixWithTxn 

func (bdb *Badger) GetByPrefixWithTxn(ctx context.Context, txn *badger.Txn,
	prefix string) (map[string][]byte, error)

func (*Badger) GetKeysByPrefix 

func (bdb *Badger) GetKeysByPrefix(ctx context.Context, prefix string) ([]string, error)

func (*Badger) GetKeysByPrefixWithTxn 

func (bdb *Badger) GetKeysByPrefixWithTxn(ctx context.Context, txn *badger.Txn,
	prefix string) ([]string, error)

func (*Badger) GetWithTxn 

func (bdb *Badger) GetWithTxn(ctx context.Context, txn *badger.Txn, key string) ([]byte, bool, error)

func (*Badger) HasChild 

func (bdb *Badger) HasChild(ctx context.Context, prefix string) bool

func (*Badger) IncrBy 

func (bdb *Badger) IncrBy(ctx context.Context, key string, delta uint64) error

func (*Badger) IncrByWithTxn 

func (bdb *Badger) IncrByWithTxn(ctx context.Context, txn *badger.Txn, key string, delta uint64) error

func (*Badger) IterateByPrefix 

func (bdb *Badger) IterateByPrefix(ctx context.Context, prefix, startCursor string, limit *int,
	process func(k, v []byte) bool) (string, error)

https://docs.hypermode.com/badger/quickstart#possible-pagination-implementation-using-prefix-scans

func (*Badger) NewTransaction 

func (bdb *Badger) NewTransaction(ctx context.Context,
	performOps func(context.Context, *badger.Txn) error) error

func (*Badger) Set 

func (bdb *Badger) Set(ctx context.Context, key string, value []byte, ttl ...time.Duration) error

func (*Badger) SetWithTxn 

func (bdb *Badger) SetWithTxn(ctx context.Context, txn *badger.Txn, key string, value []byte,
	ttl ...time.Duration) error

func (*Badger) UpdateJson 

func (bdb *Badger) UpdateJson(ctx context.Context, key string, delta map[string]any, ttl ...time.Duration) error

func (*Badger) UpdateJsonWithTxn 

func (bdb *Badger) UpdateJsonWithTxn(ctx context.Context, txn *badger.Txn, key string,
	delta map[string]any, ttl ...time.Duration) error

func (*Badger) UpdateProtobuf 

func (bdb *Badger) UpdateProtobuf(ctx context.Context, key string, delta proto.Message,
	ttl ...time.Duration) error

func (*Badger) UpdateProtobufWithTxn 

func (bdb *Badger) UpdateProtobufWithTxn(ctx context.Context, txn *badger.Txn, key string,
	delta proto.Message, ttl ...time.Duration) error

type Cluster 

type Cluster interface {
	GetID() uint64
	GetName() string
	IsNodeRemoved(id uint64) bool
}

type Config 

type Config struct {
	Name    string
	BaseDir string
	Logger  *zap.Logger

	GCInterval     time.Duration
	GCDiscardRatio float64
	BadgerOptions  badger.Options

	RaftConfig *RaftConfig
	Cluster    Cluster

	SchemaConfig *SchemaConfig

	// Hooks
	OnLeaderChange func(prevLeader, newLeader uint64)
	OnRemovedSelf  func()
}

func (*Config) PeerDialTimeout 

func (c *Config) PeerDialTimeout() time.Duration

type DBDeleteOps 

type DBDeleteOps[T OTxn] interface {
	Delete(ctx context.Context, key string) error
	DeleteWithTxn(ctx context.Context, txn T, key string) error

	DeleteByPrefix(ctx context.Context, prefix string) error
	DeleteByPrefixWithTxn(ctx context.Context, txn T, prefix string) error
}

type DBReadOps 

type DBReadOps[T OTxn] interface {
	Count(ctx context.Context, prefix string) int
	Exists(ctx context.Context, key string) bool
	HasChild(ctx context.Context, prefix string) bool

	Get(ctx context.Context, key string) ([]byte, bool, error)
	GetWithTxn(ctx context.Context, txn T, key string) ([]byte, bool, error)

	GetByPrefix(ctx context.Context, prefix string) (map[string][]byte, error)
	GetByPrefixWithTxn(ctx context.Context, txn T, prefix string) (map[string][]byte, error)

	GetKeysByPrefix(ctx context.Context, prefix string) ([]string, error)
	GetKeysByPrefixWithTxn(ctx context.Context, txn T, prefix string) ([]string, error)

	BulkGet(ctx context.Context, keys []string) (map[string][]byte, error)
	IterateByPrefix(ctx context.Context, prefix, startCursor string, limit *int, process func(k, v []byte) bool) (string, error)
}

type DBWriteOps 

type DBWriteOps[T OTxn] interface {
	DecrBy(ctx context.Context, key string, delta uint64) error
	DecrByWithTxn(ctx context.Context, txn T, key string, delta uint64) error

	IncrBy(ctx context.Context, key string, delta uint64) error
	IncrByWithTxn(ctx context.Context, txn T, key string, delta uint64) error

	Set(ctx context.Context, key string, value []byte, ttl ...time.Duration) error
	SetWithTxn(ctx context.Context, txn T, key string, value []byte, ttl ...time.Duration) error

	UpdateJson(ctx context.Context, key string, delta map[string]any, ttl ...time.Duration) error
	UpdateJsonWithTxn(ctx context.Context, txn T, key string, delta map[string]any, ttl ...time.Duration) error

	UpdateProtobuf(ctx context.Context, key string, delta proto.Message, ttl ...time.Duration) error
	UpdateProtobufWithTxn(ctx context.Context, txn T, key string, delta proto.Message, ttl ...time.Duration) error
}

type Database 

type Database[T OTxn] interface {
	DBReadOps[T]
	DBWriteOps[T]
	DBDeleteOps[T]

	GetBadger() *badger.DB
	NewTransaction(ctx context.Context, performOps func(context.Context, T) error) error
	Close(ctx context.Context)
}

type DistributedBadger 

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

func NewDistributedBadger 

func NewDistributedBadger(ctx context.Context, cfg *Config) (*DistributedBadger, error)

func (*DistributedBadger) BulkGet 

func (s *DistributedBadger) BulkGet(ctx context.Context, keys []string) (map[string][]byte, error)

func (*DistributedBadger) Close 

func (s *DistributedBadger) Close(ctx context.Context)

func (*DistributedBadger) Count 

func (s *DistributedBadger) Count(ctx context.Context, prefix string) int

func (*DistributedBadger) DecrBy 

func (s *DistributedBadger) DecrBy(ctx context.Context, key string, delta uint64) error

func (*DistributedBadger) DecrByWithTxn 

func (s *DistributedBadger) DecrByWithTxn(ctx context.Context, txn *Txn, key string, delta uint64) error

func (*DistributedBadger) Delete 

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

func (*DistributedBadger) DeleteByPrefix 

func (s *DistributedBadger) DeleteByPrefix(ctx context.Context, prefix string) error

func (*DistributedBadger) DeleteByPrefixWithTxn 

func (s *DistributedBadger) DeleteByPrefixWithTxn(ctx context.Context, txn *Txn, prefix string) error

func (*DistributedBadger) DeleteWithTxn 

func (s *DistributedBadger) DeleteWithTxn(ctx context.Context, txn *Txn, key string) error

func (*DistributedBadger) Exists 

func (s *DistributedBadger) Exists(ctx context.Context, key string) bool

func (*DistributedBadger) Get 

func (s *DistributedBadger) Get(ctx context.Context, key string) ([]byte, bool, error)

func (*DistributedBadger) GetBadger 

func (s *DistributedBadger) GetBadger() *badger.DB

func (*DistributedBadger) GetByPrefix 

func (s *DistributedBadger) GetByPrefix(ctx context.Context, prefix string) (map[string][]byte, error)

func (*DistributedBadger) GetByPrefixWithTxn 

func (s *DistributedBadger) GetByPrefixWithTxn(ctx context.Context, txn *Txn, prefix string) (map[string][]byte, error)

func (*DistributedBadger) GetKeysByPrefix 

func (s *DistributedBadger) GetKeysByPrefix(ctx context.Context, prefix string) ([]string, error)

func (*DistributedBadger) GetKeysByPrefixWithTxn 

func (s *DistributedBadger) GetKeysByPrefixWithTxn(ctx context.Context, txn *Txn, prefix string) ([]string, error)

func (*DistributedBadger) GetWithTxn 

func (s *DistributedBadger) GetWithTxn(ctx context.Context, txn *Txn, key string) ([]byte, bool, error)

func (*DistributedBadger) HasChild 

func (s *DistributedBadger) HasChild(ctx context.Context, prefix string) bool

func (*DistributedBadger) IncrBy 

func (s *DistributedBadger) IncrBy(ctx context.Context, key string, delta uint64) error

func (*DistributedBadger) IncrByWithTxn 

func (s *DistributedBadger) IncrByWithTxn(ctx context.Context, txn *Txn, key string, delta uint64) error

func (*DistributedBadger) IterateByPrefix 

func (s *DistributedBadger) IterateByPrefix(ctx context.Context, prefix, startCursor string, limit *int,
	process func(k, v []byte) bool) (string, error)

func (*DistributedBadger) NewTransaction 

func (s *DistributedBadger) NewTransaction(ctx context.Context, performOps func(context.Context, *Txn) error) error

func (*DistributedBadger) Set 

func (s *DistributedBadger) Set(ctx context.Context, key string, value []byte, ttl ...time.Duration) error

func (*DistributedBadger) SetWithTxn 

func (s *DistributedBadger) SetWithTxn(ctx context.Context, txn *Txn, key string, value []byte, ttl ...time.Duration) error

func (*DistributedBadger) UpdateJson 

func (s *DistributedBadger) UpdateJson(ctx context.Context, key string, delta map[string]any, ttl ...time.Duration) error

func (*DistributedBadger) UpdateJsonWithTxn 

func (s *DistributedBadger) UpdateJsonWithTxn(ctx context.Context, txn *Txn, key string, delta map[string]any,
	ttl ...time.Duration) error

func (*DistributedBadger) UpdateProtobuf 

func (s *DistributedBadger) UpdateProtobuf(ctx context.Context, key string, delta proto.Message,
	ttl ...time.Duration) error

func (*DistributedBadger) UpdateProtobufWithTxn 

func (s *DistributedBadger) UpdateProtobufWithTxn(ctx context.Context, txn *Txn, key string, delta proto.Message,
	ttl ...time.Duration) error

type FSM 

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

func (*FSM) Apply 

func (fsm *FSM) Apply(ctx context.Context, data []byte) error

Apply applies a committed log entry to the state machine

func (*FSM) Close 

func (fsm *FSM) Close(ctx context.Context)

func (*FSM) CreateSnapshot 

func (fsm *FSM) CreateSnapshot(ctx context.Context) (uint64, []byte, error)

func (*FSM) RestoreSnapshot 

func (fsm *FSM) RestoreSnapshot(ctx context.Context, snapshot raftpb.Snapshot) error

type Node 

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

func (*Node) IsIDRemoved 

func (node *Node) IsIDRemoved(id uint64) bool

func (*Node) IsLeader 

func (node *Node) IsLeader() bool

func (*Node) Leader 

func (node *Node) Leader() uint64

Leader returns the current leader ID

func (*Node) Process 

func (node *Node) Process(ctx context.Context, m raftpb.Message) error

func (*Node) ProposeReqNotifier 

func (node *Node) ProposeReqNotifier() chan<- propose

func (*Node) ProposeRespNotifier 

func (node *Node) ProposeRespNotifier() <-chan propose

func (*Node) ReadIndex 

func (node *Node) ReadIndex(ctx context.Context, rctx []byte) (uint64, error)

ReadIndex requests a read index for linearizable reads

func (*Node) ReportSnapshot 

func (node *Node) ReportSnapshot(id uint64, status raft.SnapshotStatus)

func (*Node) ReportUnreachable 

func (node *Node) ReportUnreachable(id uint64)

func (*Node) Start 

func (node *Node) Start(ctx context.Context, cfg *Config) error

func (*Node) Stop 

func (node *Node) Stop(ctx context.Context)

func (*Node) WaitForReadState 

func (node *Node) WaitForReadState(ctx context.Context, timeout time.Duration) (*raft.ReadState, error)

WaitForReadState waits for a read state with timeout

func (*Node) WithLeaderChangeHook 

func (node *Node) WithLeaderChangeHook(hook func(ctx context.Context, prevLeader, newLeader uint64))

func (*Node) WithRemovedSelfHook 

func (node *Node) WithRemovedSelfHook(hook func())

type OTxn 

type OTxn interface {
	*Txn | *badger.Txn
}

type RaftConfig 

type RaftConfig struct {
	raft.Config
	Nodename string
	Peers    map[uint64]string
}

type Replicate 

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

func (*Replicate) Restore 

func (r *Replicate) Restore(ctx context.Context, data []byte) error

Restore replaces the entire database with the contents of a raft snapshot.

Why DropAll before Load: DB.Load() is a merge operation — it inserts snapshot key-versions into the existing database but does NOT remove any data that is already there. For raft snapshot restoration this is incorrect: the snapshot represents the authoritative, complete state of the FSM at a specific log index. Any keys (or key-versions) that exist in the local DB but are absent from the snapshot — e.g. after a diverged log branch, a previous partial restore that crashed, or a follower catching up from behind a compaction boundary — must be erased. DropAll() clears all data while keeping the DB instance open and usable, providing the clean slate that Load() needs to faithfully reconstruct state.

func (*Replicate) TakeSnapshot 

func (r *Replicate) TakeSnapshot(ctx context.Context) (uint64, []byte, error)

TakeSnapshot creates a full in-memory backup of the database.

Why since=0: Badger's Backup(w, since) includes only key versions with commitTs >= since. The TSO's CurrentReadTs() returns a composite timestamp (physical_ms << 18 | logical) that represents the *current* (maximum) version. Passing it as `since` would include only versions committed *after* now — i.e. nothing — producing an empty snapshot. Passing 0 captures every version of every key from the beginning, which is exactly what a full raft snapshot must contain so a follower can reconstruct identical state.

Why bytes.Buffer instead of a temp file: The snapshot data must become []byte to embed in raftpb.Snapshot.Data regardless. The previous file roundtrip (Create → Backup → Close → Stat → ReadFull → Remove) added unnecessary disk I/O and latency without any benefit. Writing directly into a bytes.Buffer eliminates the intermediate disk step entirely.

type Replicator 

type Replicator interface {
	// TakeSnapshot captures a complete, consistent backup of the database and
	// returns the highest version included along with the serialised bytes.
	TakeSnapshot(ctx context.Context) (uint64, []byte, error)

	// Restore replaces the entire database with the contents of a raft snapshot.
	Restore(ctx context.Context, data []byte) error
}

func NewReplicator 

func NewReplicator(db *Badger, log *zap.Logger) (Replicator, error)

type SchemaConfig 

type SchemaConfig struct {
	Type            SchemaType
	ProtoSchemaList []*descriptorpb.FileDescriptorSet
}

type SchemaType 

type SchemaType string
var (
	SchemaTypeJson     SchemaType = "json"
	SchemaTypeProtobuf SchemaType = "protobuf"
)

type Storage 

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

func (*Storage) Close 

func (s *Storage) Close() error

func (*Storage) Compact 

func (s *Storage) Compact(index uint64) error

func (*Storage) CreateSnapshot 

func (s *Storage) CreateSnapshot(index uint64, confState raftpb.ConfState, data []byte) error

CreateSnapshot creates a new snapshot at the given index with the provided data This is typically called by the application when it wants to compact the log

func (*Storage) Entries 

func (s *Storage) Entries(lo, hi, maxSize uint64) ([]raftpb.Entry, error)

func (*Storage) Existing 

func (s *Storage) Existing() bool

func (*Storage) FirstIndex 

func (s *Storage) FirstIndex() (uint64, error)

func (*Storage) InitialState 

func (s *Storage) InitialState() (raftpb.HardState, raftpb.ConfState, error)

func (*Storage) Initialize 

func (s *Storage) Initialize(ctx context.Context) error

func (*Storage) LastIndex 

func (s *Storage) LastIndex() (uint64, error)

func (*Storage) LastSnapshotIndex 

func (s *Storage) LastSnapshotIndex() uint64

LastSnapshotIndex returns the index of the last snapshot

func (*Storage) SaveState 

func (s *Storage) SaveState(ctx context.Context, rd *raft.Ready) error

func (*Storage) Snapshot 

func (s *Storage) Snapshot() (raftpb.Snapshot, error)

func (*Storage) Term 

func (s *Storage) Term(i uint64) (uint64, error)

type TSO 

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

TSO (Timestamp Oracle) is responsible for generating unique, monotonically increasing timestamps for transactions. It maintains the current timestamp, a saved upper bound timestamp, and tracks in-flight read timestamps to manage transaction visibility and conflict detection. The TSO ensures that transactions are assigned commit timestamps in a way that respects their read timestamps and any conflicts with previously committed transactions. It also handles the persistence of the upper bound timestamp to ensure durability across restarts and provides mechanisms for cleaning up old committed transactions that are no longer relevant for conflict detection.

func (*TSO) Close 

func (tso *TSO) Close()

Close() stops the TSO's watermarks and any background goroutines. It should be called when the TSO is no longer needed to clean up resources.

func (*TSO) Commit 

func (tso *TSO) Commit(txn *Txn, fn func(ts uint64) error) error

Commit() is a helper function that generates a commit timestamp for a transaction and executes the provided function with that timestamp.

func (*TSO) CurrentReadTs 

func (tso *TSO) CurrentReadTs() uint64

CurrentReadTs() returns the current read timestamp, which is the composed value of the current timeStamp.

func (*TSO) DiscardAt 

func (tso *TSO) DiscardAt() uint64

DiscardAt() returns the timestamp until which versions can be safely discarded during compaction. It is determined by the readMark watermark, which tracks the oldest in-flight read timestamp. Any versions with timestamps less than or equal to this value can be discarded without affecting any active transactions, as they will not be visible to any ongoing reads.

func (*TSO) DoneCommit 

func (tso *TSO) DoneCommit(cts uint64)

DoneCommit() marks the given commit timestamp as done in the txnMark watermark. This should be called when a transaction that has been assigned this commit timestamp has completed its write process, allowing the TSO to advance the transaction watermark and ensure visibility of committed transactions to new reads.

func (*TSO) DoneRead 

func (tso *TSO) DoneRead(readTs uint64)

DoneRead() marks the given read timestamp as done in the readMark watermark and removes it from the readers map. This should be called when a transaction that has read at this timestamp has completed, allowing the TSO to advance the read watermark and potentially clean up old committed transactions.

func (*TSO) ReadTs 

func (tso *TSO) ReadTs(ctx context.Context) (uint64, error)

ReadTs() returns a read timestamp for a transaction. It marks the read timestamp in the readMark watermark and waits for all transactions that have been assigned a commit timestamp but have not yet completed their write process to be visible. This ensures that any transaction that reads at this timestamp will see all committed transactions up to this point, maintaining consistency.

func (*TSO) StartServing 

func (tso *TSO) StartServing(ctx context.Context) error

StartServing() starts service timestamps for transactions. It loads the last saved upper bound timestamp from the database, calibrates the TSO clock if necessary, and starts a background goroutine to periodically update the upper bound timestamp. It also initializes the read and transaction watermarks based on the current timestamp.

func (*TSO) UpdateWindow 

func (tso *TSO) UpdateWindow(ctx context.Context) error

UpdateWindow() checks if the current physical time plus a predefined upper bound window exceeds the saved upper bound timestamp. If it does, it updates the upper bound timestamp in the database and resets the saved timestamp to the new upper bound. We are using a window to reduce the frequency of updates to the database, which can be expensive. The TSO will only update the upper bound timestamp when the current time approaches the saved upper bound, ensuring that there is always a buffer of time for generating new timestamps without needing to immediately persist a new upper bound.

type Transport 

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

func NewTransport 

func NewTransport(id uint64, peers map[uint64]string, log *zap.Logger) *Transport

func (*Transport) AddPeer 

func (tr *Transport) AddPeer(ctx context.Context, id uint64, addr string)

func (*Transport) RemovePeer 

func (tr *Transport) RemovePeer(ctx context.Context, id uint64)

func (*Transport) Send 

func (tr *Transport) Send(ctx context.Context, messages []raftpb.Message)

func (*Transport) Start 

func (tr *Transport) Start(ctx context.Context, cfg *Config, node *Node, snapshotter *snap.Snapshotter)

func (*Transport) Stop 

func (tr *Transport) Stop() error

func (*Transport) UpdatePeer 

func (tr *Transport) UpdatePeer(ctx context.Context, id uint64, addr string)

type Txn 

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

func (*Txn) AddWriteKey 

func (txn *Txn) AddWriteKey(key string)

func (*Txn) Commit 

func (txn *Txn) Commit() (*types.Command, error)

func (*Txn) DecrBy 

func (txn *Txn) DecrBy(ctx context.Context, key string, delta uint64) error

func (*Txn) Delete 

func (txn *Txn) Delete(ctx context.Context, key string) error

func (*Txn) DeleteByPrefix 

func (txn *Txn) DeleteByPrefix(ctx context.Context, prefix string) error

func (*Txn) Discard 

func (txn *Txn) Discard()

func (*Txn) IncrBy 

func (txn *Txn) IncrBy(ctx context.Context, key string, delta uint64) error

func (*Txn) Set 

func (txn *Txn) Set(ctx context.Context, key string, value []byte, ttl ...time.Duration) error

func (*Txn) UpdateJson 

func (txn *Txn) UpdateJson(ctx context.Context, key string, delta map[string]any, ttl ...time.Duration) error

func (*Txn) UpdateProtobuf 

func (txn *Txn) UpdateProtobuf(ctx context.Context, key string, deltaProtoMsg proto.Message,
	ttl ...time.Duration) error

type TxnManager 

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

func (*TxnManager) BeginTxn 

func (tm *TxnManager) BeginTxn(ctx context.Context, update bool) (*Txn, error)

type Wal 

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

func (*Wal) Close 

func (w *Wal) Close() error

func (*Wal) ExistingWAL 

func (w *Wal) ExistingWAL() bool

func (*Wal) Exists 

func (w *Wal) Exists() (walExists bool, snapExists bool, err error)

Exists checks if WAL and snapshot directories exist and contain data Used to determine if this is a new node or restarting node

func (*Wal) Open 

func (w *Wal) Open(ctx context.Context) (*raftpb.Snapshot, *raftpb.HardState, []raftpb.Entry, error)

func (*Wal) Release 

func (w *Wal) Release(index uint64) error

func (*Wal) SaveSnap 

func (w *Wal) SaveSnap(snap *raftpb.Snapshot) error

func (*Wal) SaveState 

func (w *Wal) SaveState(hardState raftpb.HardState, entries []raftpb.Entry) error

func (*Wal) Sync 

func (w *Wal) Sync() error

TODO: P1: Analyze impact of Sync Because I am getting "bad file descriptor" error on calling w.wal.Close()

Source Files

View all Source files

Directories

Path Synopsis

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.