Answer mesh/state.info with a bucket's fill and caps, so a bundle's tools can say how full its state is (issue 1952)
This commit is contained in:
@@ -174,6 +174,51 @@ func (c *Conn) StateHistory(module, name, key string) ([]StateKept, error) {
|
||||
return kept, nil
|
||||
}
|
||||
|
||||
// StateInfo is how full a state's bucket is against its caps (tools-are-how-a-module-is-spoken-to): a bundle's tools
|
||||
// say fill against cap from it.
|
||||
type StateInfo struct {
|
||||
State string `json:"state"`
|
||||
Bucket string `json:"bucket"`
|
||||
// Bytes is what the bucket holds now, every kept value and marker counted; Keys how many keys it holds, a key
|
||||
// whose last value is a delete or expiry marker counted until the marker goes.
|
||||
Bytes uint64 `json:"bytes"`
|
||||
Keys uint64 `json:"keys"`
|
||||
// MaxBytes and MaxValueBytes are the bucket's caps; -1 is none.
|
||||
MaxBytes int64 `json:"max_bytes"`
|
||||
MaxValueBytes int32 `json:"max_value_bytes"`
|
||||
// History is how many values a key keeps; TTLSeconds the bucket's age for a value (0 none); KeyTTL whether a key may
|
||||
// carry its own time to live (issue 1952).
|
||||
History int64 `json:"history"`
|
||||
TTLSeconds int64 `json:"ttl_seconds"`
|
||||
KeyTTL bool `json:"key_ttl"`
|
||||
}
|
||||
|
||||
// StateInfo reads a state's bucket's fill and caps. It is a read like StateGet: served to the owner and every reader.
|
||||
func (c *Conn) StateInfo(module, name string) (*StateInfo, error) {
|
||||
s, err := c.issuedState(module, name)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
|
||||
defer cancel()
|
||||
kv, err := c.bucket(ctx, s)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
status, err := kv.Status(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
b, ok := status.(*jetstream.KeyValueBucketStatus)
|
||||
if !ok || b.StreamInfo() == nil {
|
||||
return nil, fmt.Errorf("the bus answered no stream behind %s's state %s", module, name)
|
||||
}
|
||||
info := b.StreamInfo()
|
||||
return &StateInfo{State: name, Bucket: status.Bucket(), Bytes: info.State.Bytes, Keys: uint64(info.State.NumSubjects),
|
||||
MaxBytes: info.Config.MaxBytes, MaxValueBytes: info.Config.MaxMsgSize, History: status.History(),
|
||||
TTLSeconds: int64(status.TTL() / time.Second), KeyTTL: info.Config.AllowMsgTTL && status.LimitMarkerTTL() > 0}, nil
|
||||
}
|
||||
|
||||
// StatePut writes one key, as the module, where the module keeps the state. It answers the revision.
|
||||
func (c *Conn) StatePut(module, name, key string, value json.RawMessage) (uint64, error) {
|
||||
s, err := c.writable(module, name)
|
||||
|
||||
@@ -323,7 +323,7 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an
|
||||
result = answered
|
||||
}
|
||||
case "mesh/state.get", "mesh/state.put", "mesh/state.delete", "mesh/state.keys",
|
||||
"mesh/state.create", "mesh/state.update", "mesh/state.history":
|
||||
"mesh/state.create", "mesh/state.update", "mesh/state.history", "mesh/state.info":
|
||||
var answered json.RawMessage
|
||||
if answered, err = mesh.State(strings.TrimPrefix(m.Method, "mesh/state."), m.Params); err == nil {
|
||||
result = answered
|
||||
|
||||
@@ -873,7 +873,7 @@ type stateAsked struct {
|
||||
TTLSeconds int `json:"ttl_seconds"`
|
||||
}
|
||||
|
||||
// State answers a bundle's `get`, `put`, `delete`, `keys`, `create`, `update` and `history` on its
|
||||
// State answers a bundle's `get`, `put`, `delete`, `keys`, `create`, `update`, `history` and `info` on its
|
||||
// module's state (ADR 0201, ADR 0259, ADR 0307); `create` and `put` take `ttl_seconds`, a key's own time to live,
|
||||
// whose end every watch hears as an `expire` (issue 1952). The runtime refuses, with the reason, a state the module was not issued and a write to one it only reads.
|
||||
func (b *moduleBus) State(verb string, params json.RawMessage) (json.RawMessage, error) {
|
||||
@@ -917,6 +917,13 @@ func (b *moduleBus) State(verb string, params json.RawMessage) (json.RawMessage,
|
||||
return nil, err
|
||||
}
|
||||
answer = map[string]any{}
|
||||
case "info":
|
||||
// The bucket's fill against its caps, so a bundle's tools can say how full it is.
|
||||
info, err := conn.StateInfo(b.module, asked.State)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
answer = info
|
||||
case "keys":
|
||||
keys, err := conn.StateKeys(b.module, asked.State)
|
||||
if err != nil {
|
||||
|
||||
@@ -227,6 +227,12 @@ func TestAStaleWriteIsRefusedAndWritesNothing(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// The bucket's fill against its caps, for a bundle's tools.
|
||||
if raw, err := ask(b, "info", map[string]any{"state": "wakes"}); err != nil || !strings.Contains(string(raw),
|
||||
`"bucket":"issues_wakes","bytes":0,"keys":0,"max_bytes":-1,"max_value_bytes":262144,"history":1,"ttl_seconds":0,"key_ttl":true`) {
|
||||
t.Errorf("info of a key-ttl bucket answered %s, %v", raw, err)
|
||||
}
|
||||
|
||||
// Created once, renewed whole, and its end heard: a wake with no timer.
|
||||
heard := make(chan bus.StateChange, 8)
|
||||
stop, err := b.Watch(json.RawMessage(`{"state":"wakes"}`), func(raw json.RawMessage) error {
|
||||
|
||||
Reference in New Issue
Block a user