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

Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
- API
  
  Reference documentation for the pkg.go.dev API
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

omashu

package module

Go to main page

Versions in this module

v0

v0.2.0

May 17, 2026

Changes in this version

+ const DBSubDir

+ const DefaultProposeTimeout

+ const LOGICAL_BITS

+ const MAX_LOGICAL

+ const MaxBatchSize

+ const ReadStatesTimeout

+ const RetryCount

+ const SnapSubDir

+ const WALSubDir

+ var ErrBatchTooBig = errors.New("batch size exceeds maximum limit")

+ var ErrMissingBaseDir = errors.New("baseDir is missing")

+ var ErrMissingCluster = errors.New("cluster details are missing")

+ var ErrMissingRaftConf = errors.New("raft config required for distributed db")

+ var ErrMissingSchemaConfig = errors.New("schemaConfig is missing")

+ var ErrNotLeader = errors.New("operation can only be performed on leader node")

+ var ErrProposeTimeout = errors.New("raft propose timeout")

+ var ErrUnknownOp = errors.New("unknown operation")

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

+ type Badger struct

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

+ func (bdb *Badger) IterateByPrefix(ctx context.Context, prefix, startCursor string, limit *int, ...) (string, error)

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

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

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

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

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

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

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

+ type Cluster interface

+ GetID func() uint64

+ GetName func() string

+ IsNodeRemoved func(id uint64) bool

+ type Config struct

+ BadgerOptions badger.Options

+ BaseDir string

+ Cluster Cluster

+ GCDiscardRatio float64

+ GCInterval time.Duration

+ Logger *zap.Logger

+ Name string

+ OnLeaderChange func(prevLeader, newLeader uint64)

+ OnRemovedSelf func()

+ RaftConfig *RaftConfig

+ SchemaConfig *SchemaConfig

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

+ type DBDeleteOps interface

+ Delete func(ctx context.Context, key string) error

+ DeleteByPrefix func(ctx context.Context, prefix string) error

+ DeleteByPrefixWithTxn func(ctx context.Context, txn T, prefix string) error

+ DeleteWithTxn func(ctx context.Context, txn T, key string) error

+ type DBReadOps interface

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

+ Count func(ctx context.Context, prefix string) int

+ Exists func(ctx context.Context, key string) bool

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

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

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

+ GetKeysByPrefix func(ctx context.Context, prefix string) ([]string, error)

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

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

+ HasChild func(ctx context.Context, prefix string) bool

+ IterateByPrefix func(ctx context.Context, prefix, startCursor string, limit *int, ...) (string, error)

+ type DBWriteOps interface

+ DecrBy func(ctx context.Context, key string, delta uint64) error

+ DecrByWithTxn func(ctx context.Context, txn T, key string, delta uint64) error

+ IncrBy func(ctx context.Context, key string, delta uint64) error

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

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

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

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

+ UpdateJsonWithTxn func(ctx context.Context, txn T, key string, delta map[string]any, ...) error

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

+ UpdateProtobufWithTxn func(ctx context.Context, txn T, key string, delta proto.Message, ...) error

+ type Database interface

+ Close func(ctx context.Context)

+ GetBadger func() *badger.DB

+ NewTransaction func(ctx context.Context, performOps func(context.Context, T) error) error

+ type DistributedBadger struct

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

+ func (s *DistributedBadger) IterateByPrefix(ctx context.Context, prefix, startCursor string, limit *int, ...) (string, error)

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

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

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

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

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

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

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

+ type FSM struct

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

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

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

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

+ type Node struct

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

+ func (node *Node) IsLeader() bool

+ func (node *Node) Leader() uint64

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

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

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

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

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

+ func (node *Node) ReportUnreachable(id uint64)

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

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

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

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

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

+ type OTxn interface

+ type RaftConfig struct

+ Nodename string

+ Peers map[uint64]string

+ type Replicate struct

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

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

+ type Replicator interface

+ Restore func(ctx context.Context, data []byte) error

+ TakeSnapshot func(ctx context.Context) (uint64, []byte, error)

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

+ type SchemaConfig struct

+ ProtoSchemaList []*descriptorpb.FileDescriptorSet

+ Type SchemaType

+ type SchemaType string

+ var SchemaTypeJson SchemaType = "json"

+ var SchemaTypeProtobuf SchemaType = "protobuf"

+ type Storage struct

+ func (s *Storage) Close() error

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

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

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

+ func (s *Storage) Existing() bool

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

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

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

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

+ func (s *Storage) LastSnapshotIndex() uint64

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

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

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

+ type TSO struct

+ func (tso *TSO) Close()

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

+ func (tso *TSO) CurrentReadTs() uint64

+ func (tso *TSO) DiscardAt() uint64

+ func (tso *TSO) DoneCommit(cts uint64)

+ func (tso *TSO) DoneRead(readTs uint64)

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

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

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

+ type Transport struct

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

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

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

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

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

+ func (tr *Transport) Stop() error

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

+ type Txn struct

+ func (txn *Txn) AddWriteKey(key string)

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

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

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

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

+ func (txn *Txn) Discard()

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

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

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

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

+ type TxnManager struct

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

+ type Wal struct

+ func (w *Wal) Close() error

+ func (w *Wal) ExistingWAL() bool

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

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

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

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

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

+ func (w *Wal) Sync() error