diff --git a/internal/netaddr/netaddr.go b/internal/netaddr/netaddr.go index 11bb4bd..11e1403 100644 --- a/internal/netaddr/netaddr.go +++ b/internal/netaddr/netaddr.go @@ -31,6 +31,7 @@ import ( "os" "path/filepath" "strings" + "sync" "time" "golang.org/x/sys/unix" @@ -154,31 +155,78 @@ func clientHandshake(c net.Conn, token string) error { // Listener wraps a net.Listener so Accept performs the token check for a tcp // seam. A connection that fails the check is closed and never surfaces, so // the protocol above this layer only ever sees authorized peers. +// +// A unix seam takes none of that machinery: Accept delegates straight to the +// wrapped listener, which is what it did before the token existed. type Listener struct { net.Listener addr Addr + + start sync.Once + closeOnce sync.Once + conns chan net.Conn + errc chan error // buffered 1, re-armed so every Accept sees the error + done chan struct{} } // Accept returns the next authorized connection. Unauthorized peers are // dropped and Accept keeps waiting: a bad token is a rejected stranger, not a // reason to stop serving. +// +// Each tcp handshake runs in its own goroutine rather than inline here. A peer +// that connects and then says nothing holds its greeting open for +// handshakeTimeout, and inline that peer stalls every other connection for +// five seconds — one silent stranger was enough to freeze the seam. func (l *Listener) Accept() (net.Conn, error) { + if l.addr.IsUnix() { + return l.Listener.Accept() + } + l.start.Do(func() { go l.acceptLoop() }) + select { + case c := <-l.conns: + return c, nil + case err := <-l.errc: + l.errc <- err + return nil, err + } +} + +// acceptLoop takes connections off the wrapped listener and greets each one +// concurrently. It ends on the first listener error, which every later Accept +// then reports. +func (l *Listener) acceptLoop() { for { c, err := l.Listener.Accept() if err != nil { - return nil, err + select { + case l.errc <- err: + case <-l.done: + } + return } - if l.addr.IsUnix() { - return c, nil - } - if err := serverHandshake(c, l.addr.Token); err != nil { - _ = c.Close() - continue - } - return c, nil + go l.greet(c) } } +func (l *Listener) greet(c net.Conn) { + if err := serverHandshake(c, l.addr.Token); err != nil { + _ = c.Close() + return + } + select { + case l.conns <- c: + case <-l.done: + _ = c.Close() + } +} + +// Close stops the listener and releases any connection still waiting to be +// handed to Accept. +func (l *Listener) Close() error { + l.closeOnce.Do(func() { close(l.done) }) + return l.Listener.Close() +} + // Addr reports the parsed seam address this listener was built from. func (l *Listener) SeamAddr() Addr { return l.addr } @@ -236,7 +284,7 @@ func Listen(a Addr) (*Listener, error) { if err != nil { return nil, err } - return &Listener{Listener: ln, addr: a}, nil + return wrap(ln, a), nil } if a.Token == "" { return nil, fmt.Errorf("netaddr: listen %s: tcp seam requires a token", a) @@ -245,7 +293,17 @@ func Listen(a Addr) (*Listener, error) { if err != nil { return nil, fmt.Errorf("netaddr: listen %s: %w", a, err) } - return &Listener{Listener: ln, addr: a}, nil + return wrap(ln, a), nil +} + +func wrap(ln net.Listener, a Addr) *Listener { + return &Listener{ + Listener: ln, + addr: a, + conns: make(chan net.Conn), + errc: make(chan error, 1), + done: make(chan struct{}), + } } func listenUnix(path string) (net.Listener, error) { diff --git a/internal/netaddr/netaddr_test.go b/internal/netaddr/netaddr_test.go index d0414d8..273e0bb 100644 --- a/internal/netaddr/netaddr_test.go +++ b/internal/netaddr/netaddr_test.go @@ -5,6 +5,7 @@ import ( "net" "path/filepath" "testing" + "time" ) // A scheme-less address must stay unix. Every deploy in the tree writes a bare @@ -140,6 +141,41 @@ func TestTCPUngreetedPeerDoesNotKillTheListener(t *testing.T) { } } +// A peer that connects and never speaks must not hold the seam. The greeting +// it owes is bounded by handshakeTimeout, so serving it on the accept path +// costs every later connection those five seconds. +func TestTCPSilentPeerDoesNotStallTheSeam(t *testing.T) { + ln, addr := listenLoopback(t, "s3cret") + defer ln.Close() + go echoOnce(ln) + + mute, err := net.Dial("tcp", addr.Address) + if err != nil { + t.Fatalf("mute dial: %v", err) + } + defer mute.Close() + + done := make(chan string, 1) + go func() { + c, err := Dial(addr) + if err != nil { + done <- "dial: " + err.Error() + return + } + defer c.Close() + done <- roundTrip(t, c, "still here") + }() + + select { + case got := <-done: + if got != "still here" { + t.Fatalf("got %q", got) + } + case <-time.After(handshakeTimeout / 2): + t.Fatal("a silent peer stalled the listener") + } +} + // A tcp seam with no token is a misconfiguration, and it must fail at bind // rather than serve the owner's turns to anyone who connects. func TestTCPListenRequiresToken(t *testing.T) {