From 2767278228bc3812004c16a3b55dd08fa5a7f791 Mon Sep 17 00:00:00 2001 From: mesh-admin Date: Wed, 14 Oct 2026 01:06:47 +0000 Subject: [PATCH] Say every refusal kind at once and never lose a count; bound the waiting and the busy per socket Review 12004 of mesh-host#111: refusals counted after a said one reached the journal only if the same account was refused again a minute later, and a refusal of another kind hid behind it. The first of each (account, kind) is said at once, and what is counted is said on a timer and when the socket stops. Connections waiting to be judged are bounded below the open-file limit, the busy refusals belong to one Serve, and the tests' ask has a deadline. Issue 668. Co-Authored-By: Claude Opus 5.5 --- internal/filling/filling.go | 137 ++++++++++++++++++++++++------- internal/filling/filling_test.go | 132 ++++++++++++++++++++++++++++- 2 files changed, 239 insertions(+), 30 deletions(-) diff --git a/internal/filling/filling.go b/internal/filling/filling.go index e14da2a..fc26c5e 100644 --- a/internal/filling/filling.go +++ b/internal/filling/filling.go @@ -27,8 +27,10 @@ import ( "os" "os/user" "path/filepath" + "sort" "strconv" "sync" + "sync/atomic" "time" "github.com/novox/mesh-host/internal/declaration" @@ -201,14 +203,22 @@ var ( ) // atOnce bounds the accepted connections served at the same time: one client, writing one file at a time. One over -// is refused busy. busyWaiting bounds how many busy refusals wait for their request at once; over it a busy refusal -// is written at once, as mesh-cli's socket does. Variables so a test can say. +// is refused busy. busyWaitingAtOnce bounds how many busy refusals wait for their request at once, per Serve; over it +// a busy refusal is written at once, as mesh-cli's socket does. judgingAtOnce bounds the callers judged at once, and +// judgeWaiting the connections waiting to be judged — each an open file, so far below the engine's open-file limit; +// one over is closed at once, unread. Variables so a test can say. var ( - atOnce = 4 - busyWaiting = make(chan struct{}, 8) - judgingAtOnce = 16 + atOnce = 4 + busyWaitingAtOnce = 8 + judgingAtOnce = 16 + judgeWaiting = 256 ) +// refusalsSaidEvery is how often the refusals counted and not yet said are said: each kind of refusal of each account +// is said at once the first time, and what follows it within this time is counted and said at its end, or when the +// socket stops. A variable so a test need not wait. +var refusalsSaidEvery = time.Minute + // acceptBackoff is how long the socket waits after an accept that failed before it accepts again, at first and at // most: a socket that stopped at the first such failure (too many open files, say) would stop the client's writes // until the engine restarted. Variables so a test need not wait. @@ -267,9 +277,27 @@ func Serve(ctx context.Context, socket string, w Writer, say func(string)) error // Judging reads the declaration: at most judging callers are judged at once, and the rest wait their turn — a // few milliseconds each — rather than being refused, so a flood delays the client's write and never stops it. judging := make(chan struct{}, judgingAtOnce) - refusals := &refusalWords{say: say, last: map[uint32]time.Time{}} + var waiting atomic.Int64 + busyWaiting := make(chan struct{}, busyWaitingAtOnce) + refusals := &refusalWords{say: say} + // Said once everything has stopped: what was counted is never lost with the socket. + defer refusals.flush() var wg sync.WaitGroup defer wg.Wait() + wg.Add(1) + go func() { + defer wg.Done() + tick := time.NewTicker(refusalsSaidEvery) + defer tick.Stop() + for { + select { + case <-ctx.Done(): + return + case <-tick.C: + refusals.flush() + } + } + }() backoff := time.Duration(0) for { conn, err := acceptFrom(l) @@ -292,6 +320,13 @@ func Serve(ctx context.Context, socket string, w Writer, say func(string)) error continue } backoff = 0 + if waiting.Add(1) > int64(judgeWaiting) { + // Too many waiting to be judged already: this one is closed unread, and counted. + waiting.Add(-1) + refusals.said(^uint32(0), "overflow", "too many connections are waiting to be judged; closed unread") + conn.Close() + continue + } wg.Add(1) go func() { defer wg.Done() @@ -299,6 +334,7 @@ func Serve(ctx context.Context, socket string, w Writer, say func(string)) error judging <- struct{}{} uid, ok := judgePeer(conn, w, refusals) <-judging + waiting.Add(-1) if !ok { return } @@ -309,7 +345,7 @@ func Serve(ctx context.Context, socket string, w Writer, say func(string)) error defer func() { <-slots }() answer(ctx, conn, uid, w, say) default: - busy(conn, readBy, say) + busy(conn, readBy, busyWaiting, say) } }() } @@ -318,8 +354,8 @@ func Serve(ctx context.Context, socket string, w Writer, say func(string)) error // judgePeer reads the asking account from the kernel and judges it. A caller that is not the provisioner client's // account is answered with why and closed at once, its request never read: false. func judgePeer(conn *net.UnixConn, w Writer, refusals *refusalWords) (uint32, bool) { - closeWith := func(uid uint32, why string) (uint32, bool) { - refusals.said(uid, why) + closeWith := func(uid uint32, kind, why string) (uint32, bool) { + refusals.said(uid, kind, why) _ = conn.SetWriteDeadline(time.Now().Add(refusedWithin)) _ = json.NewEncoder(conn).Encode(Refusal("", why)) conn.Close() @@ -327,44 +363,87 @@ func judgePeer(conn *net.UnixConn, w Writer, refusals *refusalWords) (uint32, bo } uid, pid, err := peerOf(conn) if err != nil { - return closeWith(^uint32(0), "the node-engine could not read which account is asking, so nothing was "+ + return closeWith(^uint32(0), "unread", "the node-engine could not read which account is asking, so nothing was "+ "written: "+err.Error()) } if err := sameNamespaces(pid); err != nil { - return closeWith(uid, "the asking process does not run in this machine's own namespaces, so nothing was "+ + return closeWith(uid, "namespace", "the asking process does not run in this machine's own namespaces, so nothing was "+ "written: "+err.Error()) } if err := w.Accepts(uid); err != nil { - return closeWith(uid, err.Error()) + return closeWith(uid, "account", err.Error()) } return uid, true } -// refusalWords says a refused caller in the journal at most once a minute for each user id, with how many more -// were refused meanwhile: a flood from another account must not flood the journal either. +// refusalWords says refused callers in the journal without letting a flood flood it: the first refusal of each kind +// for each account at once, and those that follow it counted and said by flush — on a timer and when the socket stops +// — so an account that stops still has its count said, and a refusal of another kind never hides behind one said. type refusalWords struct { - mu sync.Mutex - say func(string) - last map[uint32]time.Time - unsaid map[uint32]int + mu sync.Mutex + say func(string) + pending map[refusalKey]*refusalCount } -func (r *refusalWords) said(uid uint32, why string) { +type refusalKey struct { + uid uint32 + kind string +} + +type refusalCount struct { + n int + why string +} + +func (r *refusalWords) said(uid uint32, kind, why string) { r.mu.Lock() defer r.mu.Unlock() - if r.unsaid == nil { - r.unsaid = map[uint32]int{} + if r.pending == nil { + r.pending = map[refusalKey]*refusalCount{} } - if at, ok := r.last[uid]; ok && time.Since(at) < time.Minute { - r.unsaid[uid]++ + k := refusalKey{uid, kind} + if c, ok := r.pending[k]; ok { + c.n++ + c.why = why return } - more := "" - if n := r.unsaid[uid]; n > 0 { - more = fmt.Sprintf(" (and %d more from it since it was last said)", n) + r.pending[k] = &refusalCount{} + r.say(fmt.Sprintf("a filled file's write from %s was refused (%s): %s", who(k), kind, why)) +} + +// flush says what was counted since each was last said, and forgets every key, so the next refusal of each is said at +// once again. +func (r *refusalWords) flush() { + r.mu.Lock() + defer r.mu.Unlock() + keys := make([]refusalKey, 0, len(r.pending)) + for k := range r.pending { + keys = append(keys, k) } - r.last[uid], r.unsaid[uid] = time.Now(), 0 - r.say(fmt.Sprintf("a filled file's write from user id %d was refused%s: %s", uid, more, why)) + sort.Slice(keys, func(i, j int) bool { + if keys[i].uid != keys[j].uid { + return keys[i].uid < keys[j].uid + } + return keys[i].kind < keys[j].kind + }) + for _, k := range keys { + if c := r.pending[k]; c.n > 0 { + r.say(fmt.Sprintf("%d more writes of a filled file from %s were refused (%s) since it was last said; the last: %s", + c.n, who(k), k.kind, c.why)) + } + } + r.pending = nil +} + +// who names a caller by its user id, or says why it has none. +func who(k refusalKey) string { + switch { + case k.kind == "overflow": + return "a caller not yet judged" + case k.uid == ^uint32(0): + return "an account that could not be read" + } + return fmt.Sprintf("user id %d", k.uid) } // answer serves one connection from the provisioner client's account. @@ -396,7 +475,7 @@ func answer(ctx context.Context, conn *net.UnixConn, uid uint32, w Writer, say f // busy answers the client's connection over atOnce: its request read and dropped first, briefly, so the client reads // the answer rather than a broken pipe — while no more than busyWaiting such refusals wait at once; over that, at once. -func busy(conn *net.UnixConn, readBy time.Time, say func(string)) { +func busy(conn *net.UnixConn, readBy time.Time, busyWaiting chan struct{}, say func(string)) { select { case busyWaiting <- struct{}{}: by := time.Now().Add(refusedWithin) diff --git a/internal/filling/filling_test.go b/internal/filling/filling_test.go index 2c198ad..0e7cff3 100644 --- a/internal/filling/filling_test.go +++ b/internal/filling/filling_test.go @@ -224,6 +224,8 @@ func ask(t *testing.T, socket string, req Request) Answer { t.Fatal(err) } defer conn.Close() + // Bounded: a socket that never answers fails the test instead of hanging it. + _ = conn.SetDeadline(time.Now().Add(5 * time.Second)) if err := json.NewEncoder(conn).Encode(req); err != nil { t.Fatal(err) } @@ -327,7 +329,7 @@ func TestAFloodFromAnotherAccountDoesNotStopTheClient(t *testing.T) { } } -func busyWaitingCap() int { return cap(busyWaiting) } +func busyWaitingCap() int { return busyWaitingAtOnce } // A client that connects and says nothing holds one slot only until within, and another write goes on meanwhile. func TestASilentClientDoesNotHoldTheSocket(t *testing.T) { @@ -429,3 +431,131 @@ func TestTheSocketsDirectoryIsOpenedToEveryAccount(t *testing.T) { t.Errorf("the socket's directory is %o", info.Mode().Perm()) } } + +// The first refusal of each kind for each account is said at once; what follows within the time is counted, and the +// count is said when the time is up, even when the account has stopped. +func TestRefusalsAreSaidOnceAndTheirCountIsNeverLost(t *testing.T) { + var mu sync.Mutex + var said []string + r := &refusalWords{say: func(s string) { mu.Lock(); said = append(said, s); mu.Unlock() }} + for i := 0; i < 3; i++ { + r.said(1000, "account", "not the provisioner client's account") + } + // Another kind from the same account is said at once, not hidden behind the first. + r.said(1000, "namespace", "another namespace") + if len(said) != 2 || !strings.Contains(said[0], "(account)") || !strings.Contains(said[1], "(namespace)") { + t.Fatalf("said at once: %q", said) + } + // The account stops; the time is up; the two counted are said. + r.flush() + if len(said) != 3 || !strings.Contains(said[2], "2 more") || !strings.Contains(said[2], "user id 1000") { + t.Fatalf("after the time: %q", said) + } + // And the next refusal is said at once again. + r.said(1000, "account", "not the provisioner client's account") + if len(said) != 4 { + t.Fatalf("after the count was said: %q", said) + } +} + +// refusingSocket serves a socket that refuses this test's account, with counted refusals said every every. It returns +// how to be refused n times, everything said so far, and how to stop it. +func refusingSocket(t *testing.T, every time.Duration) (func(int), func() string, func()) { + t.Helper() + was := refusalsSaidEvery + refusalsSaidEvery = every + dir, err := os.MkdirTemp("", "fill") + if err != nil { + t.Fatal(err) + } + socket := filepath.Join(dir, "fill.sock") + ctx, cancel := context.WithCancel(context.Background()) + var mu sync.Mutex + var said []string + all := func() string { mu.Lock(); defer mu.Unlock(); return strings.Join(said, "\n") } + stopped := make(chan error) + go func() { + stopped <- Serve(ctx, socket, &fixed{uid: uint32(os.Getuid()) + 1}, + func(s string) { mu.Lock(); said = append(said, s); mu.Unlock() }) + }() + for i := 0; i < 100; i++ { + if _, err := os.Stat(socket); err == nil { + break + } + time.Sleep(10 * time.Millisecond) + } + refuse := func(n int) { + for i := 0; i < n; i++ { + conn, err := net.Dial("unix", socket) + if err != nil { + t.Fatal(err) + } + _ = conn.SetDeadline(time.Now().Add(5 * time.Second)) + var a Answer + _ = json.NewDecoder(conn).Decode(&a) + conn.Close() + } + } + var once sync.Once + stop := func() { + once.Do(func() { + cancel() + if err := <-stopped; err != nil { + t.Error(err) + } + refusalsSaidEvery = was + os.RemoveAll(dir) + }) + } + t.Cleanup(stop) + return refuse, all, stop +} + +// Refusals counted while the socket serves are said on its timer, though the account has stopped asking. +func TestTheSocketSaysItsCountedRefusalsOnATimer(t *testing.T) { + refuse, all, _ := refusingSocket(t, 200*time.Millisecond) + refuse(3) + // "more writes", not a number: a tick may fall between two refusals and split the count. + deadline := time.Now().Add(3 * time.Second) + for !strings.Contains(all(), "more writes") && time.Now().Before(deadline) { + time.Sleep(20 * time.Millisecond) + } + if !strings.Contains(all(), "more writes") { + t.Fatalf("the counted refusals were not said on the timer: %s", all()) + } +} + +// Refusals counted when the socket stops are said as it stops. +func TestTheSocketSaysItsCountedRefusalsWhenItStops(t *testing.T) { + refuse, all, stop := refusingSocket(t, time.Hour) + refuse(5) + if strings.Contains(all(), "4 more") { + t.Fatalf("said before the socket stopped: %s", all()) + } + stop() + if !strings.Contains(all(), "4 more") { + t.Errorf("the refusals counted when the socket stopped were lost: %s", all()) + } +} + +// A connection over the bound of those waiting to be judged is closed unread, and said. +func TestAConnectionOverTheWaitingBoundIsClosedUnread(t *testing.T) { + was := judgeWaiting + judgeWaiting = 0 + t.Cleanup(func() { judgeWaiting = was }) + w := &fixed{uid: uint32(os.Getuid())} + socket, said := serve(t, w) + conn, err := net.Dial("unix", socket) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + _ = conn.SetDeadline(time.Now().Add(5 * time.Second)) + var a Answer + if err := json.NewDecoder(conn).Decode(&a); err == nil { + t.Fatalf("answered over the bound: %+v", a) + } + if len(w.writes) != 0 || !strings.Contains(strings.Join(said(), "\n"), "(overflow)") { + t.Errorf("writes %v, said %q", w.writes, said()) + } +}