Merge pull request 'nats: say whether a user's grant is in force on the running server (review of hq issue 353)' (#166) from feat/353-the-bus-says-a-grant-is-in-force into main
This commit was merged in pull request #166.
This commit is contained in:
@@ -29,7 +29,7 @@ func limitOf(a map[string]any, fallback int) int {
|
||||
return fallback
|
||||
}
|
||||
|
||||
func tools(monitor Monitor, users func(context.Context) ([]byte, error)) []stdio.Tool {
|
||||
func tools(monitor Monitor, users func(context.Context) ([]byte, error), written func(context.Context) (time.Time, error)) []stdio.Tool {
|
||||
get := func(path string) ([]byte, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
|
||||
defer cancel()
|
||||
@@ -123,20 +123,28 @@ func tools(monitor Monitor, users func(context.Context) ([]byte, error)) []stdio
|
||||
return Users(c, arg(a, "user")), nil
|
||||
}},
|
||||
{Name: "nats_user_can",
|
||||
Description: "Whether a bus user may publish or subscribe to a subject, and the grant that allows it — the question behind a request that timed out because the bus refused it.",
|
||||
Description: "Whether a bus user may publish or subscribe to a subject, and the grant that allows it — the question behind a request that timed out because the bus refused it. Also whether the running server holds it: `in_force` is true only when the user list on the machine allows it and the server reloaded its configuration after that list was written (novox/hq issue 353: a list sent is not a list the bus enforces until it reloads).",
|
||||
Input: map[string]any{"user": str("the user exactly, e.g. g14.node-tools"), "action": str("publish or subscribe"), "subject": str("a concrete subject, e.g. $KV.claude-code_servers.all.x")},
|
||||
Run: func(a map[string]any) (any, error) {
|
||||
c, err := conf()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return UserCan(c, arg(a, "user"), arg(a, "action"), arg(a, "subject"))
|
||||
answer, err := UserCan(c, arg(a, "user"), arg(a, "action"), arg(a, "subject"))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
|
||||
defer cancel()
|
||||
at, werr := written(ctx)
|
||||
varz, verr := get("/varz")
|
||||
return InForce(answer, at, werr, varz, verr), nil
|
||||
}},
|
||||
}
|
||||
}
|
||||
|
||||
func main() {
|
||||
if err := stdio.Serve("", tools(LiveMonitor(), ReadUsers)); err != nil {
|
||||
if err := stdio.Serve("", tools(LiveMonitor(), ReadUsers, UsersWritten)); err != nil {
|
||||
fmt.Fprintln(os.Stderr, err)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
@@ -65,3 +65,24 @@ func ReadUsers(ctx context.Context) ([]byte, error) {
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// UsersWritten is when the composed user list the server includes was last written, read in the container
|
||||
// to the nanosecond (`stat -c %y`): what the server's last reload must come after for the list to be in force.
|
||||
func UsersWritten(ctx context.Context) (time.Time, error) {
|
||||
container := env("MESH_NATS_CONTAINER", "mesh-broker-nats")
|
||||
path := env("MESH_NATS_USERS_FILE", "/etc/nats/accounts.conf")
|
||||
out, err := exec.CommandContext(ctx, "docker", "exec", container, "stat", "-c", "%y", path).Output()
|
||||
if err != nil {
|
||||
return time.Time{}, fmt.Errorf("when the bus's user list was written could not be read inside %s: %w", container, err)
|
||||
}
|
||||
return ParseWritten(string(out))
|
||||
}
|
||||
|
||||
// ParseWritten reads `stat -c %y`: "2026-10-09 17:38:13.422211788 +0000".
|
||||
func ParseWritten(s string) (time.Time, error) {
|
||||
t, err := time.Parse("2006-01-02 15:04:05.999999999 -0700", strings.TrimSpace(s))
|
||||
if err != nil {
|
||||
return time.Time{}, fmt.Errorf("when the bus's user list was written is not a time: %q", strings.TrimSpace(s))
|
||||
}
|
||||
return t, nil
|
||||
}
|
||||
|
||||
@@ -536,3 +536,39 @@ func UserCan(conf []byte, user, action, subject string) (map[string]any, error)
|
||||
}
|
||||
return nil, fmt.Errorf("the bus has no user %q; nats_users lists them", user)
|
||||
}
|
||||
|
||||
// InForce adds to a UserCan answer whether the running server holds what the file says (novox/hq issue 353):
|
||||
// `in_force` only when the grant is allowed in the file and the server's last configuration load
|
||||
// (`config_load_time` of /varz) came after the file was last written. Anything not known is not in force, said
|
||||
// in `why`: an asker waiting for its grant must not take a list sent for a list the bus enforces.
|
||||
func InForce(answer map[string]any, written time.Time, writtenErr error, varz []byte, varzErr error) map[string]any {
|
||||
answer["in_force"] = false
|
||||
switch {
|
||||
case writtenErr != nil:
|
||||
answer["why"] = writtenErr.Error()
|
||||
return answer
|
||||
case varzErr != nil:
|
||||
answer["why"] = "the server's last configuration load could not be read: " + varzErr.Error()
|
||||
return answer
|
||||
}
|
||||
var v struct {
|
||||
ConfigLoadTime time.Time `json:"config_load_time"`
|
||||
}
|
||||
if err := json.Unmarshal(varz, &v); err != nil || v.ConfigLoadTime.IsZero() {
|
||||
answer["why"] = "the server does not say when it last loaded its configuration"
|
||||
return answer
|
||||
}
|
||||
answer["file_written"] = written.UTC().Format(time.RFC3339Nano)
|
||||
answer["server_loaded"] = v.ConfigLoadTime.UTC().Format(time.RFC3339Nano)
|
||||
loaded := !v.ConfigLoadTime.Before(written)
|
||||
allowed, _ := answer["allowed"].(bool)
|
||||
switch {
|
||||
case !allowed:
|
||||
answer["why"] = "the user list on the machine does not allow it"
|
||||
case !loaded:
|
||||
answer["why"] = "the user list on the machine allows it, and the server has not reloaded since it was written"
|
||||
default:
|
||||
answer["in_force"] = true
|
||||
}
|
||||
return answer
|
||||
}
|
||||
|
||||
@@ -2,6 +2,8 @@ package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -170,3 +172,39 @@ func TestClosedConnectionsArePerUserAndNameAndSayWhichWereDropped(t *testing.T)
|
||||
t.Errorf("narrowed: %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
// novox/hq issue 353: a grant written to the user list is in force only once the server reloaded after the
|
||||
// write; anything not known is not in force.
|
||||
func TestAGrantIsInForceOnlyWhenTheServerReloadedAfterTheListWasWritten(t *testing.T) {
|
||||
written, err := ParseWritten("2026-10-09 17:55:56.422211788 +0200\n")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
varz := func(at string) []byte { return []byte(`{"config_load_time":"` + at + `"}`) }
|
||||
allowed := func() map[string]any { return map[string]any{"allowed": true} }
|
||||
cases := []struct {
|
||||
name string
|
||||
answer map[string]any
|
||||
werr error
|
||||
varz []byte
|
||||
verr error
|
||||
want bool
|
||||
says string
|
||||
}{
|
||||
{"reloaded after the write", allowed(), nil, varz("2026-10-09T15:56:05.1Z"), nil, true, ""},
|
||||
{"reloaded before the write", allowed(), nil, varz("2026-10-09T15:55:56.4Z"), nil, false, "has not reloaded"},
|
||||
{"not allowed in the file", map[string]any{"allowed": false}, nil, varz("2026-10-09T15:56:05Z"), nil, false, "does not allow"},
|
||||
{"the write time unknown", allowed(), errors.New("no container"), varz("2026-10-09T15:56:05Z"), nil, false, "no container"},
|
||||
{"the server unknown", allowed(), nil, nil, errors.New("refused"), false, "could not be read"},
|
||||
{"a server that says no load time", allowed(), nil, []byte(`{}`), nil, false, "does not say"},
|
||||
}
|
||||
for _, c := range cases {
|
||||
got := InForce(c.answer, written, c.werr, c.varz, c.verr)
|
||||
if got["in_force"] != c.want || (c.says != "" && !strings.Contains(fmt.Sprint(got["why"]), c.says)) {
|
||||
t.Errorf("%s: %v", c.name, got)
|
||||
}
|
||||
}
|
||||
if _, err := ParseWritten("yesterday"); err == nil {
|
||||
t.Error("a write time that is not one was read")
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user