The runtime's channel carries 16 MiB a message, so a value over the 8 MiB a put takes is split over objects (Ref.Parts). Builds: design 47 G §57, §138 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
304 lines
8.7 KiB
Go
304 lines
8.7 KiB
Go
package stateobjects
|
|
|
|
import (
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/kvlimit"
|
|
)
|
|
|
|
// ErrChanged is a write refused because the key was written by another writer first; nothing was written.
|
|
var ErrChanged = errors.New("the key was written by another writer first")
|
|
|
|
// Entry is a key's current value and its revision.
|
|
type Entry struct {
|
|
Value []byte
|
|
Revision uint64
|
|
}
|
|
|
|
// KV is the state's key-value side, as a bundle reaches it through its runtime or a process through its own JetStream.
|
|
type KV interface {
|
|
// Get is the key's current entry, or nil when it has none.
|
|
Get(key string) (*Entry, error)
|
|
// Create writes the key only when it has none, Update only while it is at revision; either refuses as ErrChanged.
|
|
Create(key string, value []byte) error
|
|
Update(key string, value []byte, revision uint64) error
|
|
Delete(key string) error
|
|
// Keys is every key with a value.
|
|
Keys() ([]string, error)
|
|
}
|
|
|
|
// Object is one object an object store holds.
|
|
type Object struct {
|
|
Name string
|
|
Size int64
|
|
Modified time.Time
|
|
}
|
|
|
|
// Objects is the object store beside the state.
|
|
type Objects interface {
|
|
// MostBytes is the largest object it takes in one put, or zero for no limit: a value over it is split over objects.
|
|
MostBytes() int
|
|
Put(name string, data []byte) error
|
|
Get(name string) ([]byte, error)
|
|
Delete(name string) error
|
|
List() ([]Object, error)
|
|
}
|
|
|
|
// Values is a state whose values may be of any size: a value that fits the bucket is kept in its key, a larger one in
|
|
// the object store beside it with its reference in the key (ADR 0666 §16). Store is the object store's name as a reader
|
|
// asks for it, written into every reference.
|
|
type Values struct {
|
|
KV KV
|
|
Objects Objects
|
|
Store string
|
|
// Limit is the most a key's value may hold, headers included; zero is kvlimit.ModuleStateValueBytes.
|
|
Limit int
|
|
}
|
|
|
|
func (v Values) inlineMost() int {
|
|
limit := v.Limit
|
|
if limit <= 0 {
|
|
limit = kvlimit.ModuleStateValueBytes
|
|
}
|
|
return limit - kvlimit.HeaderRoom
|
|
}
|
|
|
|
func digestOf(data []byte) string {
|
|
sum := sha256.Sum256(data)
|
|
return hex.EncodeToString(sum[:])
|
|
}
|
|
|
|
// objectName is a large value's object, `<key>.<digest>`: the same value is the same object, a changed one another.
|
|
func objectName(key, digest string) string { return key + "." + digest }
|
|
|
|
// Put writes the value as JSON under key: inline up to the limit, else as an object `<key>.<sha256>` with its reference
|
|
// in the key. A replace removes the object it replaced in the same act; a write another writer won first is ErrChanged
|
|
// and leaves what it found. A value that is itself exactly a reference is refused.
|
|
func (v Values) Put(key string, value any) error {
|
|
data, err := json.Marshal(value)
|
|
if err != nil {
|
|
return fmt.Errorf("the value for %s: %w", key, err)
|
|
}
|
|
if _, isRef := ReadRef(data); isRef {
|
|
return fmt.Errorf("%s: %w", key, errLooksLikeRef)
|
|
}
|
|
old, err := v.KV.Get(key)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var oldRef Ref
|
|
var hadRef bool
|
|
if old != nil {
|
|
oldRef, hadRef = ReadRef(old.Value)
|
|
}
|
|
body, object := data, ""
|
|
var newRef Ref
|
|
if len(data) > v.inlineMost() {
|
|
ref := Ref{Store: v.Store, Object: objectName(key, digestOf(data)), Size: int64(len(data)), Digest: digestOf(data), Form: FormVersion}
|
|
object = ref.Object
|
|
if !(hadRef && oldRef.Object == object) {
|
|
parts, err := putObject(v.Objects, object, data)
|
|
if err != nil {
|
|
return fmt.Errorf("the object for %s: %w", key, err)
|
|
}
|
|
ref.Parts = parts
|
|
} else {
|
|
ref.Parts = oldRef.Parts
|
|
}
|
|
body = Wrap(ref)
|
|
newRef = ref
|
|
}
|
|
if old == nil {
|
|
err = v.KV.Create(key, body)
|
|
} else {
|
|
err = v.KV.Update(key, body, old.Revision)
|
|
}
|
|
if err != nil {
|
|
if object != "" && !(hadRef && oldRef.Object == object) {
|
|
_ = deleteObject(v.Objects, newRef)
|
|
}
|
|
return err
|
|
}
|
|
if hadRef && oldRef.Object != object {
|
|
if err := deleteObject(v.Objects, oldRef); err != nil {
|
|
return fmt.Errorf("%s is written; the object it replaced, %s, could not be removed: %w", key, oldRef.Object, err)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Get reads the key's value into into and says whether the key has one. A value kept as an object is fetched and
|
|
// checked against its reference's size and digest.
|
|
func (v Values) Get(key string, into any) (bool, error) {
|
|
e, err := v.KV.Get(key)
|
|
if err != nil || e == nil {
|
|
return false, err
|
|
}
|
|
data := e.Value
|
|
if ref, ok := ReadRef(e.Value); ok {
|
|
if data, err = getObject(v.Objects, ref); err != nil {
|
|
return false, fmt.Errorf("the object %s of %s: %w", ref.Object, key, err)
|
|
}
|
|
if int64(len(data)) != ref.Size || digestOf(data) != ref.Digest {
|
|
return false, fmt.Errorf("the object %s of %s is not the one its reference names (size or digest differ)", ref.Object, key)
|
|
}
|
|
}
|
|
return true, json.Unmarshal(data, into)
|
|
}
|
|
|
|
// Delete removes the key. Its object stays, listed as key-gone, until the drop tool removes it on a person's word
|
|
// (ADR 0666 §16, ADR 0595 §4).
|
|
func (v Values) Delete(key string) error { return v.KV.Delete(key) }
|
|
|
|
// Kept is one object of a state's store as the reading tool shows it.
|
|
type Kept struct {
|
|
Name string `json:"name"`
|
|
Size int64 `json:"size"`
|
|
Modified time.Time `json:"modified,omitzero"`
|
|
// Key is the key whose value names it; for an object whose key is gone, the key it was written for.
|
|
Key string `json:"key"`
|
|
// KeyGone says no key names the object any longer.
|
|
KeyGone bool `json:"key-gone,omitempty"`
|
|
}
|
|
|
|
// Kept lists every object of the store with the key naming it, or key-gone, by name.
|
|
func (v Values) Kept() ([]Kept, error) {
|
|
live, err := v.live()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
objects, err := v.Objects.List()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out := make([]Kept, 0, len(objects))
|
|
for _, o := range objects {
|
|
k := Kept{Name: o.Name, Size: o.Size, Modified: o.Modified}
|
|
if key, ok := live[o.Name]; ok {
|
|
k.Key = key
|
|
} else {
|
|
k.KeyGone = true
|
|
k.Key = keyOf(o.Name)
|
|
}
|
|
out = append(out, k)
|
|
}
|
|
sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })
|
|
return out, nil
|
|
}
|
|
|
|
// live is the objects some key names now, by object name to key.
|
|
func (v Values) live() (map[string]string, error) {
|
|
keys, err := v.KV.Keys()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
live := map[string]string{}
|
|
for _, key := range keys {
|
|
e, err := v.KV.Get(key)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if e == nil {
|
|
continue
|
|
}
|
|
if ref, ok := ReadRef(e.Value); ok {
|
|
for _, name := range ref.names() {
|
|
live[name] = key
|
|
}
|
|
}
|
|
}
|
|
return live, nil
|
|
}
|
|
|
|
// DropKeyGone removes the objects whose key is gone, and only those, answering their names. It is the act of the drop
|
|
// tool; a live object, one a key names, is never removed.
|
|
func (v Values) DropKeyGone() ([]string, error) {
|
|
kept, err := v.Kept()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var dropped []string
|
|
for _, k := range kept {
|
|
if !k.KeyGone {
|
|
continue
|
|
}
|
|
if err := v.Objects.Delete(k.Name); err != nil {
|
|
return dropped, fmt.Errorf("the object %s: %w", k.Name, err)
|
|
}
|
|
dropped = append(dropped, k.Name)
|
|
}
|
|
return dropped, nil
|
|
}
|
|
|
|
// names are the objects a reference stands for.
|
|
func (r Ref) names() []string {
|
|
if r.Parts <= 1 {
|
|
return []string{r.Object}
|
|
}
|
|
out := make([]string, r.Parts)
|
|
for i := range out {
|
|
out[i] = fmt.Sprintf("%s.%d", r.Object, i)
|
|
}
|
|
return out
|
|
}
|
|
|
|
// keyOf is the key an object `<key>.<digest>` or `<key>.<digest>.<n>` was written for.
|
|
func keyOf(object string) string {
|
|
isDigest := func(s string) bool { _, err := hex.DecodeString(s); return len(s) == 64 && err == nil }
|
|
parts := strings.Split(object, ".")
|
|
if n := len(parts); n >= 3 && isDigest(parts[n-2]) {
|
|
return strings.Join(parts[:n-2], ".")
|
|
}
|
|
if n := len(parts); n >= 2 && isDigest(parts[n-1]) {
|
|
return strings.Join(parts[:n-1], ".")
|
|
}
|
|
return object
|
|
}
|
|
|
|
// putObject stores data as one object, or over `<name>.<n>` when the store takes none so large; it answers the parts
|
|
// (zero for one object) and removes what it stored when a part fails.
|
|
func putObject(o Objects, name string, data []byte) (int, error) {
|
|
most := o.MostBytes()
|
|
if most <= 0 || len(data) <= most {
|
|
return 0, o.Put(name, data)
|
|
}
|
|
ref := Ref{Object: name, Parts: (len(data) + most - 1) / most}
|
|
for i, part := range ref.names() {
|
|
from, to := i*most, min((i+1)*most, len(data))
|
|
if err := o.Put(part, data[from:to]); err != nil {
|
|
_ = deleteObject(o, Ref{Object: name, Parts: i})
|
|
return 0, err
|
|
}
|
|
}
|
|
return ref.Parts, nil
|
|
}
|
|
|
|
func getObject(o Objects, ref Ref) ([]byte, error) {
|
|
var out []byte
|
|
for _, name := range ref.names() {
|
|
d, err := o.Get(name)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, d...)
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func deleteObject(o Objects, ref Ref) error {
|
|
var first error
|
|
for _, name := range ref.names() {
|
|
if err := o.Delete(name); err != nil && first == nil {
|
|
first = err
|
|
}
|
|
}
|
|
return first
|
|
}
|