shutdown: close the sockets, or the database never gets sealed
mavend seals its encrypted database in `defer st.Close()` when run() returns. It had not returned since 2026-07-21. Every restart since then decrypted the same eleven-day-old ciphertext and rolled back everything written in between: the Telegram nudge that kept firing was a fact being un-written on each boot. The goroutine dump named it. main → srv.Close() → ipc.(*Server).Close → wg.Wait(), waiting on per-connection goroutines parked in readFrame. Close shut the listener and nothing else, so the idle persistent sockets held by mavweb, mavpoll, mavcaldav and mavmaild blocked shutdown forever. `docker compose stop -t 60` spent the whole sixty seconds and then took a SIGKILL. So: track the accepted conns and close them, in ipc and in voice, which had the identical defect. Bound all three waits — the two per-server ones and the worker wait in main — because the seal matters more than any single in-flight call. A dropped RPC costs one reply; a missed seal costs a session. The regression test leaves a client connected and idle, which is the case the old tests avoided by closing the client first. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TrVSBKe3RFDF4fGYKWYQnX
This commit is contained in:
@@ -57,8 +57,20 @@ type Server struct {
|
||||
ln net.Listener
|
||||
wg sync.WaitGroup
|
||||
done chan struct{}
|
||||
|
||||
// Accepted conns, tracked so Close can shut them. Same defect as
|
||||
// ipc/server.go had: closing only the listener leaves every idle client
|
||||
// parked in readFrame, wg.Wait never returns, and the daemon dies to
|
||||
// SIGKILL without sealing the database.
|
||||
connMu sync.Mutex
|
||||
conns map[net.Conn]struct{}
|
||||
}
|
||||
|
||||
// closeGrace — how long Close waits for in-flight dispatches before dropping
|
||||
// them. A push-to-talk turn can be mid-inference; abandoning one costs a reply,
|
||||
// hanging costs every write since the last clean shutdown.
|
||||
const closeGrace = 3 * time.Second
|
||||
|
||||
// NewServer builds a Server bound to addr (e.g. "127.0.0.1:9100" for a
|
||||
// local-only smoke; production: a wg-tunnel address). handler is the
|
||||
// reactive handler; sessions is shared with the voicesink (the daemon
|
||||
@@ -111,8 +123,10 @@ func (s *Server) Serve() error {
|
||||
}
|
||||
}
|
||||
s.wg.Add(1)
|
||||
s.trackConn(c)
|
||||
go func(c net.Conn) {
|
||||
defer s.wg.Done()
|
||||
defer s.untrackConn(c)
|
||||
s.serveConn(c)
|
||||
}(c)
|
||||
}
|
||||
@@ -212,10 +226,65 @@ func (s *Server) Close() error {
|
||||
if s.ln != nil {
|
||||
err = s.ln.Close()
|
||||
}
|
||||
s.wg.Wait()
|
||||
// Close the accepted conns too, or a client that is merely idle keeps
|
||||
// serveConn blocked in readFrame forever.
|
||||
s.closeConns()
|
||||
if !waitTimeout(&s.wg, closeGrace) {
|
||||
log.Printf("voice: %d connection(s) still busy after %s, closing anyway", s.liveConns(), closeGrace)
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func (s *Server) trackConn(c net.Conn) {
|
||||
s.connMu.Lock()
|
||||
defer s.connMu.Unlock()
|
||||
if s.conns == nil {
|
||||
s.conns = make(map[net.Conn]struct{})
|
||||
}
|
||||
s.conns[c] = struct{}{}
|
||||
}
|
||||
|
||||
func (s *Server) untrackConn(c net.Conn) {
|
||||
s.connMu.Lock()
|
||||
defer s.connMu.Unlock()
|
||||
delete(s.conns, c)
|
||||
}
|
||||
|
||||
func (s *Server) liveConns() int {
|
||||
s.connMu.Lock()
|
||||
defer s.connMu.Unlock()
|
||||
return len(s.conns)
|
||||
}
|
||||
|
||||
// closeConns unblocks every parked reader. serveConn's own defer closes the
|
||||
// conn again; a second Close on a net.Conn is a harmless error.
|
||||
func (s *Server) closeConns() {
|
||||
s.connMu.Lock()
|
||||
conns := make([]net.Conn, 0, len(s.conns))
|
||||
for c := range s.conns {
|
||||
conns = append(conns, c)
|
||||
}
|
||||
s.connMu.Unlock()
|
||||
for _, c := range conns {
|
||||
_ = c.Close()
|
||||
}
|
||||
}
|
||||
|
||||
// waitTimeout waits on wg, but not forever. Reports whether it finished.
|
||||
func waitTimeout(wg *sync.WaitGroup, d time.Duration) bool {
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
wg.Wait()
|
||||
close(done)
|
||||
}()
|
||||
select {
|
||||
case <-done:
|
||||
return true
|
||||
case <-time.After(d):
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func unmarshalParams(raw json.RawMessage, v any) error {
|
||||
if len(raw) == 0 {
|
||||
raw = []byte("null")
|
||||
|
||||
Reference in New Issue
Block a user