clouddb

README

CloudDB

Another attempt at building a fault tolerant storage system.

This provides no guarantee of ACID compliance, no transaction isolation and such.

It can store JSON objects only.

• Primary keys will need to be globally unique, using for example uuid.NewRandom()
• JSON object contains a number of values, always start with '{' (no support for root to be anything else than an object)
• Search index keys can be specified and will be replicated globally
• Any update will be transfromed into a delta JSON and replicated as delta only (TODO)
• All updates will have a timestamp and be applied based on timestamp value
• A global log of all updates is stored and kept for a limited amount of time (TODO cleanup of old records)
• All checkpoints also have a log count value that represents the size of work so far and allows detecting missing logs
• Local data is stored in a single leveldb in order to allow consistent snapshots

Database consistency:

• Log entries sorted by nanotime timestamp
• Checkpoints will include number of known log entries
• We produce one checkpoint per 24 hours by default in order to ensure that even if a server goes down it can come back up

Leveldb prefixes

leveldb prefixes are fixed at 3 bytes for ease of storage.

• idx+key = id (indices)
• log+logid = log (journal, indexed by timestamp + hash)
• dat+id = data (data)
• nfo+id = version (16 bytes current record version)
• kdt+id = keys (key data, 32 bits length of key followed by key, repeated for each key)
• typ+type_name = type data
• chk+id = checkpoint

Journal

Key: log + timestamp(16bytes) + hash(sha256)

checkpoints

• We perform one checkpoint every 24 hours
• Checkpoint is rounded timestamp + number of log entries in the past 24 hours + xor of log keys
• We only keep 100 latest checkpoints (3+ months)

snapshots

• We perform a snapshot within a few minutes of a checkpoint being reached
• The snapshot will include all the changes up to the checkpoint, and some more
• Upon starting from zero, a node will load the latest snapshot and download it, then apply subsequent log except for log already included in the snapshot since considered already applied
• Log entries prior to the snapshot date will not be included in the snapshot

Documentation

Index

• Constants
• Variables
• func ApplyUpdate(obj any, upd any) (any, error)
• func ComputeDelta(old, now any) (any, error)
• func RegisterType(t *Type)
• type DB
  • func New(name string, rpc RPC) (*DB, error)
  • func (d *DB) Close() error
  • func (d *DB) DebugDump(w io.Writer)
  • func (d *DB) Delete(id []byte) error
  • func (d *DB) Get(id []byte, target any) error
  • func (d *DB) GetRaw(id []byte) (json.RawMessage, error)
  • func (d *DB) GetStatus() Status
  • func (d *DB) Has(id []byte) (bool, error)
  • func (d *DB) SearchFirst(typ string, search map[string]any, target any) ([]byte, error)
  • func (d *DB) SearchFirstRaw(typ string, search map[string]any) ([]byte, json.RawMessage, error)
  • func (d *DB) Set(id []byte, val any) error
  • func (d *DB) String() string
  • func (d *DB) WaitReady()
• type RPC
• type Status
  • func (s Status) String() string
• type Type
  • func (t *Type) Hash() []byte
• type TypeKey

Constants

const (
	RecordSet    dblogType = iota // set the full value of a record
	RecordDelete                  // delete a record
)

const (
	PktGetInfo     = 0x00 // GetInfo()
	PktGetLogIds   = 0x01 // GetLogs(<peer>, <epoch>)
	PktFetchLog    = 0x02 // FetchLog(<key>) uses no response packet id, response is always inline
	PktGetInfoResp = 0x80 // GetInfo()|Response
	PktLogPush     = 0x40 // pushed log entry
	PktLogIdsPush  = 0x41 // LogIdPush(<peer>, ...<logid>)

	ResponseFlag = 0x80 // response flag
)

Variables

var (
	ErrKeyConflict = errors.New("record's key conflicts with an existing record")
)

Functions

func ApplyUpdate

func ApplyUpdate(obj any, upd any) (any, error)

ApplyUpdate will update the value of obj with upd, merging changes if both are objects. In most other cases upd will be returned as is. Both values should be of the same type, or nil.

func ComputeDelta

func ComputeDelta(old, now any) (any, error)

ComputeDelta compares two objects (of type map[string]any) and returns an object that contains all the changes found in now while skipping any value that might be equal.

func RegisterType

func RegisterType(t *Type)

Types

type DB

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

func New

func New(name string, rpc RPC) (*DB, error)

func (*DB) Close

func (d *DB) Close() error

func (*DB) DebugDump

func (d *DB) DebugDump(w io.Writer)

DebugDump will dump the whole database to log.DefaultLogger

func (*DB) Delete

func (d *DB) Delete(id []byte) error

Delete globally removes an entry from the database

func (*DB) Get

func (d *DB) Get(id []byte, target any) error

Get will load and apply the object for the given id to target

func (*DB) GetRaw

func (d *DB) GetRaw(id []byte) (json.RawMessage, error)

GetRaw returns the json value for a given id

func (*DB) GetStatus

func (d *DB) GetStatus() Status

GetStatus returns the current status of this database instance

func (*DB) Has

func (d *DB) Has(id []byte) (bool, error)

Has will check if a given key exists

func (*DB) SearchFirst

func (d *DB) SearchFirst(typ string, search map[string]any, target any) ([]byte, error)

SearchFirst will find the first record matching the search params and set target search must match an existing key in the provided type

func (*DB) SearchFirstRaw

func (d *DB) SearchFirstRaw(typ string, search map[string]any) ([]byte, json.RawMessage, error)

SearchFirstRaw will find the first record matching the search params and return its json value search must match an existing key in the provided type

func (*DB) Set

func (d *DB) Set(id []byte, val any) error

Set will update a given record in database. id must be unique in the whole db, even across types

func (*DB) String

func (d *DB) String() string

func (*DB) WaitReady

func (d *DB) WaitReady()

WaitReady waits for the database instance to be ready, or does nothing if it is already ready

type RPC

type RPC interface {
	// All will send a given data object to all other RPC instances on the fleet
	// and will collect responses
	All(ctx context.Context, data []byte) ([]any, error)

	// Broadcast will do the same as All but will not wait for responses
	Broadcast(ctx context.Context, data []byte) error

	// Request will send a given object to a specific peer and return the response
	Request(ctx context.Context, id string, data []byte) ([]byte, error)

	// Send will send a given buffer to a specific peer and ignore the response
	Send(ctx context.Context, id string, data []byte) error

	// Self will return the id of the local peer, can be used for other instances
	// to contact here with Send().
	Self() string

	// ListPeers returns a list of connected peers
	ListOnlinePeers() []string

	// CountAllPeers return the number of known connected or offline peers
	CountAllPeers() int

	// Connect connects this RPC instance incoming events to a given function
	// that will be called each time an event is received.
	Connect(cb func(context.Context, []byte) ([]byte, error))
}

RPC interface is standardized

type Status

type Status int

const (
	Initializing Status = iota
	Syncing
	Ready
	Error
)

func (Status) String

func (s Status) String() string

type Type

type Type struct {
	Name    string     `json:"name"`
	Keys    []*TypeKey `json:"keys"`
	Version int        `json:"version"` // version for the definition, can be set to zero
}

Type is a registered index type used in records. A record of a given type will inherit the passed keys and be indexed based on it. Some keys will always exist, such as to ensure objects can be referenced at all times.

Types will be stored in the database to ensure records can be indexed properly on first try.

func (*Type) Hash

func (t *Type) Hash() []byte

type TypeKey

type TypeKey struct {
	Name   string   `json:"name"`
	Fields []string `json:"fields"`
	Method string   `json:"method"` // one of: utf8, int, binary
	Unique bool     `json:"unique,omitempty"`
}