// Package streamcap bounds how many live streams anonymous readers may
// hold open. Each stream keeps a goroutine and, for chat, a broker
// consumer alive for as long as the connection lasts, and an anonymous
// reader can open them without signing in, so nothing else stops one
// client from holding as many as it likes.
package streamcap
import (
"net/netip"
"sync"
)
const (
// PerAddress is how many anonymous streams one address may hold.
// A reader with several tabs open on several channels needs a few,
// and a household or office behind one address needs a few more;
// a client past eight is opening streams rather than reading them.
PerAddress = 8
// Total is how many anonymous streams the whole process may hold.
// It bounds the goroutines and broker consumers anonymous readers
// can pin at once, with room for a busy public channel, while
// staying small beside what the broker and the process can carry.
Total = 256
// RetryAfter is the Retry-After a refused stream is sent, in
// seconds. A slot frees when one of the reader's own streams ends,
// usually a tab closing, so a short fixed wait is honest; there is
// no queue position to report and no point in a client retrying
// in a tight loop.
RetryAfter = "30"
)
// Anonymous is the process-wide budget every anonymous stream draws
// on. It is one budget rather than one per surface because the cost it
// bounds, goroutines and broker consumers, is the process's, however
// the stream was reached.
var Anonymous = New(PerAddress, Total)
// Limiter counts open streams per address and in total.
type Limiter struct {
perAddr, total int
mu sync.Mutex
n int
byAddr map[string]int
}
// New returns a limiter admitting perAddr streams from one address and
// total streams overall.
func New(perAddr, total int) *Limiter {
return &Limiter{perAddr: perAddr, total: total, byAddr: map[string]int{}}
}
// Acquire admits one stream from addr, or reports that addr or the
// process is at its cap. addr is the caller's to decide, by the
// forge's trusted-proxy rule; this package never reads a request, so
// it cannot grow a second rule of its own. The caller must call release when the stream
// ends; calling it more than once frees the slot only once.
func (l *Limiter) Acquire(addr string) (release func(), ok bool) {
addr = key(addr)
l.mu.Lock()
defer l.mu.Unlock()
if l.n >= l.total || l.byAddr[addr] >= l.perAddr {
return nil, false
}
l.n++
l.byAddr[addr]++
var once sync.Once
return func() {
once.Do(func() {
l.mu.Lock()
defer l.mu.Unlock()
l.n--
if l.byAddr[addr]--; l.byAddr[addr] <= 0 {
delete(l.byAddr, addr)
}
})
}, true
}
// key is the unit an address is counted as. An IPv6 client is normally
// handed a whole /64 and can take a fresh address from it for every
// stream, so counting exact IPv6 addresses would let one /64 fill the
// process-wide Total and shut every other anonymous reader out. The /64
// is counted as one reader instead. IPv4, and IPv4 carried in IPv6, stay
// exact, since an IPv4 neighbour is usually somebody else. Anything that
// is not an address is counted as given.
func key(addr string) string {
a, err := netip.ParseAddr(addr)
if err != nil || !a.Is6() || a.Is4In6() {
return addr
}
return netip.PrefixFrom(a.WithZone(""), 64).Masked().String()
}