package bus import ( "context" "encoding/json" "errors" "fmt" "regexp" "sort" "strings" "sync" "time" "github.com/nats-io/nats.go/jetstream" ) // A module's state on the bus (novox/hq ADR 0202): key-value buckets the controller creates from what // the module declared, and issues to each assignment in its membership by the name the module uses — // its own state by the local name, another module's as `.`. // // **The runtime keeps each module to its own buckets.** One account per machine carries every module // on it, so the bus enforces only the union; and a write the bus refuses reaches the writer as a // timeout, not a refusal (measured, novox/hq research 024). So what a module may reach is decided here, // from its membership, and refused with the reason before anything is sent. // StateIssued is one bucket an assignment may reach (ADR 0202). type StateIssued struct { Name string `json:"name"` Bucket string `json:"bucket"` Writes bool `json:"writes,omitempty"` } // StateEntry is one key's current value. type StateEntry struct { Key string `json:"key"` Value json.RawMessage `json:"value"` Revision uint64 `json:"revision"` } // StateChange is one change a watch delivers: a key put or deleted. type StateChange struct { State string `json:"state"` Key string `json:"key"` Op string `json:"op"` Value json.RawMessage `json:"value,omitempty"` Revision uint64 `json:"revision"` // Current is true for a value that was there when the watch began, false for a change since. Current bool `json:"current"` } // StateTimeout bounds one state operation on the bus. var StateTimeout = 10 * time.Second // stateKey is a key the bus can hold: letters, digits and `-/_=.`, no leading or trailing dot. A // module's convention for naming a machine in a key (`all.`, `.`) fits it. var stateKey = regexp.MustCompile(`^[-/_=a-zA-Z0-9]+(\.[-/_=a-zA-Z0-9]+)*$`) // issuedState is the bucket a module may reach by a name, and whether it may write it — or the reason // it may not reach it at all. func (c *Conn) issuedState(module, name string) (StateIssued, error) { m := c.Membership(module) if m == nil { return StateIssued{}, fmt.Errorf("%s has no membership issued on %s yet, so no state of it is reachable "+ "until the mesh issues one (novox/hq ADR 0202)", module, c.node) } var names []string for _, s := range m.State { if s.Name == name { return s, nil } names = append(names, s.Name) } sort.Strings(names) issued := "none" if len(names) > 0 { issued = strings.Join(names, ", ") } return StateIssued{}, fmt.Errorf("%s keeps and reads no state called %q: it declares what it keeps under "+ "`state` and what it reads under `reads` as ., and was issued: %s (novox/hq ADR 0202)", module, name, issued) } func (c *Conn) bucket(ctx context.Context, s StateIssued) (jetstream.KeyValue, error) { js, err := jetstream.New(c.nc) if err != nil { return nil, err } kv, err := js.KeyValue(ctx, s.Bucket) if err != nil { if errors.Is(err, jetstream.ErrBucketNotFound) { return nil, fmt.Errorf("the state %q is issued and its bucket is not on the bus yet: the controller "+ "creates it from the catalogue on its next raise", s.Name) } return nil, err } return kv, nil } func checkKey(key string) error { if !stateKey.MatchString(key) { return fmt.Errorf("%q is not a key the bus can hold: letters, digits and -/_=, in dot-separated "+ "names", key) } return nil } // StateGet is one key's current value, or nil when it has none. func (c *Conn) StateGet(module, name, key string) (*StateEntry, error) { s, err := c.issuedState(module, name) if err != nil { return nil, err } if err := checkKey(key); 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 } e, err := kv.Get(ctx, key) if errors.Is(err, jetstream.ErrKeyNotFound) { return nil, nil } if err != nil { return nil, err } return &StateEntry{Key: e.Key(), Value: valueOf(e.Value()), Revision: e.Revision()}, 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) if err != nil { return 0, err } if err := checkKey(key); err != nil { return 0, err } if len(value) == 0 || !json.Valid(value) { return 0, fmt.Errorf("a state value is JSON") } if field := credentialField(value); field != "" { return 0, fmt.Errorf("%s's %s.%s carries a field %q, which names a credential: no secret is kept in state, "+ "sealed or not — a bucket is a stream, and a machine joining a year later reads it whole. Name the "+ "secret and fetch it on request/reply (novox/hq ADR 0202, design 32 §10)", module, name, key, field) } ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) defer cancel() kv, err := c.bucket(ctx, s) if err != nil { return 0, err } return kv.Put(ctx, key, value) } // StateDelete removes one key, as the module, where the module keeps the state. A key that was not // there is not an error: what is asked for is that it is gone. func (c *Conn) StateDelete(module, name, key string) error { s, err := c.writable(module, name) if err != nil { return err } if err := checkKey(key); err != nil { return err } ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) defer cancel() kv, err := c.bucket(ctx, s) if err != nil { return err } return kv.Delete(ctx, key) } // StateKeys is every key with a current value, sorted. func (c *Conn) StateKeys(module, name string) ([]string, 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 } lister, err := kv.ListKeys(ctx) if err != nil { return nil, err } defer func() { _ = lister.Stop() }() keys := []string{} for k := range lister.Keys() { keys = append(keys, k) } sort.Strings(keys) return keys, nil } func (c *Conn) writable(module, name string) (StateIssued, error) { s, err := c.issuedState(module, name) if err != nil { return s, err } if !s.Writes { owner := name if dot := strings.LastIndex(name, "."); dot > 0 { owner = name[:dot] } return s, fmt.Errorf("%s reads %s and does not keep it: only %s's own instances write it (novox/hq ADR 0202)", module, name, owner) } return s, nil } // StateWatch hands deliver the current value of every key matching the pattern — none that is // deleted — and then every change, in order (ADR 0202). It returns once the current values are // delivered; deliver is called from one goroutine, one change at a time, and an error from it is said // and the watch goes on: state is not a queue, and the next change, or the next start, reads it again. // The pattern is a key, with `*` for one name and `**` for the rest; empty is every key. func (c *Conn) StateWatch(module, name, pattern string, deliver func(StateChange) error) (stop func(), err error) { s, err := c.issuedState(module, name) if err != nil { return nil, err } filter := ">" if pattern != "" { parts := strings.Split(pattern, ".") for i, p := range parts { if p == "**" { if i != len(parts)-1 { return nil, fmt.Errorf("%q: `**` stands for the rest of a key, so it comes last", pattern) } parts[i] = ">" continue } if p != "*" && !stateKey.MatchString(p) { return nil, fmt.Errorf("%q is not a key pattern: names, `*` for one and `**` for the rest", pattern) } } filter = strings.Join(parts, ".") } ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) kv, err := c.bucket(ctx, s) cancel() if err != nil { return nil, err } watching, stopWatching := context.WithCancel(context.Background()) w, err := kv.Watch(watching, filter) if err != nil { stopWatching() return nil, err } var once sync.Once stop = func() { once.Do(func() { _ = w.Stop() stopWatching() }) } current := make(chan struct{}) go func() { initial := true for e := range w.Updates() { if e == nil { // The end of what was there when the watch began. if initial { initial = false close(current) } continue } change := StateChange{State: name, Key: e.Key(), Revision: e.Revision(), Current: initial} switch e.Operation() { case jetstream.KeyValuePut: change.Op = "put" change.Value = valueOf(e.Value()) default: // A deletion among the current values is a key that is not there: not handed over // (measured, research 024 — the server sends its marker among the initial values). if initial { continue } change.Op = "delete" } if err := deliver(change); err != nil { c.Logf("[mesh-tools] %s did not take %s %s of %s: %v; its next change, or its next start, reads it again", module, change.Op, change.Key, name, err) } } if initial { close(current) } }() select { case <-current: return stop, nil case <-time.After(StateTimeout): stop() return nil, fmt.Errorf("the current values of %s did not arrive in %s", name, StateTimeout) } } func valueOf(raw []byte) json.RawMessage { if len(raw) == 0 || !json.Valid(raw) { b, _ := json.Marshal(string(raw)) return b } return json.RawMessage(raw) } // credentialEndings are the ends of a field name that say its value is a credential. A guard against // the ordinary mistake, not a determined one: a sealed value is plain text to anything inspecting it, // so the rule is checked where it can be and said to be partial (ADR 0202). var credentialEndings = []string{"password", "passwd", "secret", "token", "credential", "credentials", "authorization", "apikey", "privatekey", "accesskey", "cookie"} // credentialField is the first field anywhere in a JSON value whose name says it is a credential, or "". func credentialField(value json.RawMessage) string { var v any if json.Unmarshal(value, &v) != nil { return "" } var walk func(any) string walk = func(v any) string { switch t := v.(type) { case map[string]any: keys := make([]string, 0, len(t)) for k := range t { keys = append(keys, k) } sort.Strings(keys) for _, k := range keys { norm := strings.NewReplacer("-", "", "_", "", " ", "").Replace(strings.ToLower(k)) for _, end := range credentialEndings { if strings.HasSuffix(norm, end) { return k } } if found := walk(t[k]); found != "" { return found } } case []any: for _, x := range t { if found := walk(x); found != "" { return found } } } return "" } return walk(v) }