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 <noreply@anthropic.com>

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)

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())
+	}
+}