fix(server): drain live games on SIGTERM and honour a second signal

The wsapi server was built on the signal context, so SIGTERM cancelled
every room before StartDraining ran and the restart notice never went
out. Build it on a background context, release the signal context as
soon as it fires, and wait a bounded moment after Shutdown so the notice
reaches open sockets.

Also: Hard prefers the slower loss in lost positions; resigning out of
turn no longer settles a pending dead end on the spot; a self-closing
<ref> with a slash in its name no longer swallows definition text; the
store enforces builder_version; both listeners get an IdleTimeout; the
real-corpus ladder is reproducible from its seed; a -healthcheck flag
probes /healthz for the container health check.
This commit is contained in:
tiennm99 committed 2026-09-29 20:33:16 +07:00
1 parent d3eb13e36e
commit 7fb724d29e
14 files changed
+600 -88

No files matched your search

+4 -2
View File
@@ -36,8 +36,10 @@ import (
)
// builderVer changes whenever the meta table's contract does, so two databases
// with different provenance rows never claim the same builder.
const builderVer = "5"
// with different provenance rows never claim the same builder. It is the
// store's own constant, so the version the builder writes and the one Open
// requires cannot drift apart.
const builderVer = dictionary.RequiredBuilderVersion
// minMeaningCoverage is the share of words a dump build must carry a meaning
// for. The 2026-09-01 dump measured well above it; the floor exists to catch a
+3 -3
View File
@@ -27,7 +27,7 @@ var vietnameseOnsets = []string{
// vietnameseCodas is the complete inventory of syllable-final consonants and
// offglides.
var vietnameseCodas = []string{
"ngh", "ng", "nh", "ch",
"ng", "nh", "ch",
"c", "m", "n", "p", "t", "i", "o", "u", "y",
}
@@ -36,7 +36,7 @@ var vietnameseCodas = []string{
var vietnameseNuclei = []string{
"uye", "uya", "uyu", "oai", "oay", "uoi", "uou", "ieu", "yeu", "uai", "uay",
"ai", "ao", "au", "ay", "eo", "eu", "ia", "ie", "iu", "oa", "oe", "oi", "oo",
"ua", "ue", "ui", "uo", "uu", "uy", "ya", "ye", "yu", "ao", "eu",
"ua", "ue", "ui", "uo", "uu", "uy", "ya", "ye", "yu",
"a", "e", "i", "o", "u", "y",
}
@@ -66,7 +66,7 @@ func stripDiacritics(syllable string) string {
func isCombiningMark(r rune) bool {
// The ranges Vietnamese actually uses; cheaper and tighter than a full
// unicode.Is(unicode.Mn, r) for this data.
return (r >= 0x0300 && r <= 0x036F) || r == 0x031B
return r >= 0x0300 && r <= 0x036F
}
// isVietnameseSyllable reports whether a syllable fits Vietnamese phonotactics.
+6 -3
View File
@@ -126,9 +126,12 @@ var (
langnameVi = regexp.MustCompile(`^\{\{langname\|vi\}\}$`)
htmlComment = regexp.MustCompile(`(?s)<!--.*?-->`)
refElement = regexp.MustCompile(`(?s)<ref\b[^>/]*/>|<ref\b[^>]*>.*?</ref>`)
anyTag = regexp.MustCompile(`</?[A-Za-z][^>]*>`)
spaces = regexp.MustCompile(`\s+`)
// refElement's self-closing branch reads quoted attribute values whole, so
// a slash inside one (name="a/b") does not stop it from recognising the
// element and letting the paired branch run on to the next </ref>.
refElement = regexp.MustCompile(`(?s)<ref\b(?:[^>"]|"[^"]*")*/>|<ref\b[^>]*>.*?</ref>`)
anyTag = regexp.MustCompile(`</?[A-Za-z][^>]*>`)
spaces = regexp.MustCompile(`\s+`)
)
// isLegacyHeading reports whether a {{-code-}} is a heading inside a language
@@ -42,6 +42,9 @@ func TestStripWikitext(t *testing.T) {
{"ref mid-sentence and a lone ref",
"Một loài [[cá]]<ref>Từ điển</ref> nước ngọt<ref name=\"a\" />.",
"Một loài cá nước ngọt."},
{"self-closing ref whose attribute holds a slash keeps the text after it",
"Một từ.<ref name=\"a/b\"/> Nghĩa thêm <ref>x</ref> cuối.",
"Một từ. Nghĩa thêm cuối."},
{"comment, bold, italic, entities",
"'''Rất''' ''nhanh''<!-- todo -->&nbsp;và&amp;mạnh.",
"Rất nhanh và&mạnh."},
+133 -23
View File
@@ -6,7 +6,10 @@ import (
"context"
"errors"
"expvar"
"flag"
"fmt"
"log/slog"
"net"
"net/http"
"os"
"os/signal"
@@ -35,6 +38,24 @@ const (
// timeout by more than a blink, cheap enough to poll at all — the
// alternative is a channel the hub would need to fan out to every room.
drainPollInterval = 200 * time.Millisecond
// shutdownFlush is how long the process lingers after telling every
// session the server is restarting. Each session writes that notice from
// its own goroutine, and http.Server.Shutdown does not wait for hijacked
// WebSocket connections, so returning at once could exit before the
// notice reached a socket. wsapi exposes no live-session count to wait on,
// so this is a short fixed bound.
shutdownFlush = 2 * time.Second
// idleTimeout closes idle keep-alive HTTP connections, which would
// otherwise be held forever outside every connection cap. It is
// deliberately the only connection deadline: ReadTimeout and WriteTimeout
// stay on a hijacked connection and would drop every WebSocket game after
// that interval.
idleTimeout = 120 * time.Second
// healthcheckTimeout bounds the -healthcheck request.
healthcheckTimeout = 3 * time.Second
)
// version is the build the process is running. The default here is what
@@ -56,11 +77,23 @@ type config struct {
maxConnectionsPerIP int
debugAddr string
drainTimeout time.Duration
flushWait time.Duration
}
func main() {
slog.SetDefault(slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelInfo})))
healthcheck := flag.Bool("healthcheck", false,
"probe GET /healthz on the configured address (NOITU_ADDR) and exit 0 if it answers 200, 1 otherwise")
flag.Parse()
if *healthcheck {
if err := checkHealth(env("NOITU_ADDR", defaultAddr), healthcheckTimeout); err != nil {
fmt.Fprintln(os.Stderr, "healthcheck failed:", err)
os.Exit(1)
}
return
}
if err := run(); err != nil {
slog.Error("server exited", "err", err)
os.Exit(1)
@@ -86,10 +119,33 @@ func run() error {
"license", store.License(),
)
ln, err := net.Listen("tcp", cfg.addr)
if err != nil {
return err
}
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()
api := wsapi.NewServer(ctx, store, wsapi.Config{
return serve(ctx, stop, cfg, store, ln)
}
// serve runs the server on ln until ctx is cancelled — by a signal in
// production — or the listener fails, then drains and shuts down.
//
// ctx is only the "stop now" trigger. The wsapi server is built on its own
// background context, because a room derives its lifetime from the context it
// is given: handing it the signal context would cancel every room and session
// at the moment of the signal, before the drain could keep a game alive or
// tell a player the server is restarting. api.Shutdown cancels that context
// itself once the drain is over.
//
// release is called as soon as ctx fires. In production it is the signal
// context's stop, which restores the default signal behaviour so a second
// SIGTERM or SIGINT during a long drain terminates the process instead of
// being swallowed. It may be nil.
func serve(ctx context.Context, release func(), cfg config, dict wsapi.Dictionary, ln net.Listener) error {
api := wsapi.NewServer(context.Background(), dict, wsapi.Config{
TurnLimit: cfg.turnLimit,
GraceFor: cfg.grace,
AllowedOrigins: cfg.allowedOrigins,
@@ -101,18 +157,13 @@ func run() error {
Version: version,
})
srv := &http.Server{
Addr: cfg.addr,
Handler: api,
ReadHeaderTimeout: 10 * time.Second,
}
srv := newHTTPServer(cfg.addr, api)
debugSrv := newDebugServer(cfg.debugAddr)
errc := make(chan error, 1)
go func() {
slog.Info("listening", "addr", cfg.addr, "turn_limit", cfg.turnLimit, "web_dir", cfg.webDir, "version", version)
if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
slog.Info("listening", "addr", ln.Addr().String(), "turn_limit", cfg.turnLimit, "web_dir", cfg.webDir, "version", version)
if err := srv.Serve(ln); err != nil && !errors.Is(err, http.ErrServerClosed) {
errc <- err
}
}()
@@ -127,28 +178,86 @@ func run() error {
select {
case err := <-errc:
api.Shutdown()
_ = shutdownServer(debugSrv)
return err
case <-ctx.Done():
}
if release != nil {
release()
}
// Draining is the first half of shutdown: stop seating new rooms and flip
// /readyz unhealthy, so a load balancer stops sending this instance new
// traffic while the games it already has finish on their own turn clock.
// A creator refused during this window gets server_restarting rather than
// server_full — the room is not coming back, unlike a full one.
slog.Info("draining", "rooms", api.RoomCount(), "live_games", api.LiveGameCount())
api.StartDraining()
waitForGamesToFinish(api, cfg.drainTimeout)
// Tell players why before the sockets go, rather than dropping them and
// leaving the UI to guess.
slog.Info("shutting down", "rooms", api.RoomCount(), "live_games", api.LiveGameCount())
api.Shutdown()
drainAndShutdown(api, cfg.drainTimeout, cfg.flushWait)
_ = shutdownServer(debugSrv)
return shutdownServer(srv)
}
// lifecycle is the slice of *wsapi.Server the shutdown sequence drives,
// narrowed so the ordering can be tested against a recorder.
type lifecycle interface {
gameCounter
RoomCount() int
StartDraining()
Shutdown()
}
// drainAndShutdown runs the shutdown sequence in its required order.
//
// Draining comes first: stop seating new rooms and flip /readyz unhealthy, so
// a load balancer stops sending this instance new traffic while the games it
// already has finish on their own turn clock. A creator refused during this
// window gets server_restarting rather than server_full — the room is not
// coming back, unlike a full one. Only then are players told why the sockets
// are going, and flush gives those notices time to reach them.
func drainAndShutdown(api lifecycle, drainTimeout, flush time.Duration) {
slog.Info("draining", "rooms", api.RoomCount(), "live_games", api.LiveGameCount())
api.StartDraining()
waitForGamesToFinish(api, drainTimeout)
slog.Info("shutting down", "rooms", api.RoomCount(), "live_games", api.LiveGameCount())
api.Shutdown()
if flush > 0 {
time.Sleep(flush)
}
}
// newHTTPServer builds a listener — the public one and the debug one alike. Only ReadHeaderTimeout and
// IdleTimeout are set: see idleTimeout for why the other deadlines are not.
func newHTTPServer(addr string, handler http.Handler) *http.Server {
return &http.Server{
Addr: addr,
Handler: handler,
ReadHeaderTimeout: 10 * time.Second,
IdleTimeout: idleTimeout,
}
}
// checkHealth GETs /healthz on the address the server listens on and reports
// anything but a 200 as an error. It exists because the container image is
// distroless: no curl or wget for a container health check to shell out to.
func checkHealth(addr string, timeout time.Duration) error {
host, port, err := net.SplitHostPort(addr)
if err != nil {
return fmt.Errorf("address %q: %w", addr, err)
}
// A wildcard listen address is not something to dial; loopback reaches it.
if ip := net.ParseIP(host); host == "" || (ip != nil && ip.IsUnspecified()) {
host = "127.0.0.1"
}
client := &http.Client{Timeout: timeout}
resp, err := client.Get("http://" + net.JoinHostPort(host, port) + "/healthz")
if err != nil {
return err
}
defer func() { _ = resp.Body.Close() }()
if resp.StatusCode != http.StatusOK {
return fmt.Errorf("/healthz answered %s", resp.Status)
}
return nil
}
// shutdownServer stops srv gracefully, giving in-flight requests up to
// shutdownGrace. A nil srv — the debug listener when it is not configured —
// is a no-op.
@@ -171,7 +280,7 @@ func newDebugServer(addr string) *http.Server {
}
mux := http.NewServeMux()
mux.Handle("/debug/vars", expvar.Handler())
return &http.Server{Addr: addr, Handler: mux, ReadHeaderTimeout: 10 * time.Second}
return newHTTPServer(addr, mux)
}
// gameCounter is the drain loop's only dependency on *wsapi.Server, narrowed
@@ -224,6 +333,7 @@ func loadConfig() config {
maxConnectionsPerIP: envInt("NOITU_MAX_CONNECTIONS_PER_IP", 0),
debugAddr: env("NOITU_DEBUG_ADDR", ""),
drainTimeout: envNonNegDuration("NOITU_DRAIN_TIMEOUT", 0),
flushWait: shutdownFlush,
}
}
+296
View File
@@ -0,0 +1,296 @@
package main
import (
"context"
"database/sql"
"net"
"net/http"
"net/http/httptest"
"path/filepath"
"strings"
"sync"
"testing"
"time"
"github.com/coder/websocket"
"google.golang.org/protobuf/proto"
noituv1 "github.com/tiennm99dev/noitu/server/gen/noitu/v1"
"github.com/tiennm99dev/noitu/server/internal/dictionary"
"github.com/tiennm99dev/noitu/server/internal/wsapi"
_ "modernc.org/sqlite"
)
// chainStore opens a three-word dictionary through the real store, so serve is
// exercised against the same Dictionary implementation production uses.
func chainStore(t *testing.T) *dictionary.Store {
t.Helper()
path := filepath.Join(t.TempDir(), "noitu.db")
db, err := sql.Open("sqlite", "file:"+path)
if err != nil {
t.Fatal(err)
}
defer func() { _ = db.Close() }()
_, err = db.Exec(`
CREATE TABLE words (word TEXT PRIMARY KEY, first TEXT NOT NULL, last TEXT NOT NULL, syllables INTEGER NOT NULL) WITHOUT ROWID;
CREATE INDEX idx_words_first ON words(first);
CREATE TABLE syllables (syllable TEXT PRIMARY KEY, out_degree INTEGER NOT NULL) WITHOUT ROWID;
CREATE TABLE aliases (variant TEXT PRIMARY KEY, canonical TEXT NOT NULL) WITHOUT ROWID;
CREATE TABLE meanings (word TEXT NOT NULL, ord INTEGER NOT NULL, pos TEXT NOT NULL, gloss TEXT NOT NULL, PRIMARY KEY (word, ord)) WITHOUT ROWID;
CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT NOT NULL);
INSERT INTO meta VALUES ('builder_version','` + dictionary.RequiredBuilderVersion + `'),('source_license','CC BY-SA 4.0'),('word_count','3'),('meaning_count','0');
INSERT INTO words VALUES ('a b','a','b',2),('b c','b','c',2),('c d','c','d',2);
INSERT INTO syllables VALUES ('a',1),('b',1),('c',1),('d',0);`)
if err != nil {
t.Fatal(err)
}
store, err := dictionary.Open(path)
if err != nil {
t.Fatalf("Open: %v", err)
}
return store
}
// openingAtA lets the test choose the opening word: the store's own picker
// wants a syllable with many continuations, which a three-word chain lacks.
type openingAtA struct{ *dictionary.Store }
func (openingAtA) RandomOpeningWord(int) (string, error) { return "a b", nil }
// The deploy path: a signal must start the drain while the game is still
// alive, keep waiting for it up to the drain timeout, tell the player the
// server is restarting, and only then let the process exit.
func TestServeDrainsLiveGamesBeforeShuttingDown(t *testing.T) {
ln, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
addr := ln.Addr().String()
const drain = 1500 * time.Millisecond
cfg := config{
addr: addr,
turnLimit: 30 * time.Second,
grace: 30 * time.Second,
drainTimeout: drain,
flushWait: 100 * time.Millisecond,
}
// The context stands in for the signal context.
sig, signal := context.WithCancel(context.Background())
defer signal()
released := make(chan struct{})
done := make(chan error, 1)
go func() {
done <- serve(sig, func() { close(released) }, cfg, openingAtA{chainStore(t)}, ln)
}()
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
defer cancel()
conn := dialWhenUp(t, ctx, addr)
defer func() { _ = conn.CloseNow() }()
write := func(m *noituv1.ClientMessage) {
t.Helper()
raw, err := proto.Marshal(m)
if err != nil {
t.Fatal(err)
}
if err := conn.Write(ctx, websocket.MessageBinary, raw); err != nil {
t.Fatalf("write: %v", err)
}
}
read := func() (*noituv1.ServerMessage, error) {
_, raw, err := conn.Read(ctx)
if err != nil {
return nil, err
}
var m noituv1.ServerMessage
return &m, proto.Unmarshal(raw, &m)
}
await := func(match func(*noituv1.ServerMessage) bool) *noituv1.ServerMessage {
t.Helper()
for range 20 {
m, err := read()
if err != nil {
t.Fatalf("read before the expected message: %v", err)
}
if match(m) {
return m
}
}
t.Fatal("expected message never arrived")
return nil
}
write(&noituv1.ClientMessage{Payload: &noituv1.ClientMessage_Hello{Hello: &noituv1.Hello{
ProtocolVersion: wsapi.ProtocolVersion, Nickname: "Người thử",
}}})
await(func(m *noituv1.ServerMessage) bool { return m.GetWelcome() != nil })
write(&noituv1.ClientMessage{Payload: &noituv1.ClientMessage_StartBotGame{
StartBotGame: &noituv1.StartBotGame{Difficulty: noituv1.Difficulty_DIFFICULTY_EASY},
}})
await(func(m *noituv1.ServerMessage) bool { return m.GetGameStarted() != nil })
begin := time.Now()
signal()
select {
case <-released:
case <-time.After(5 * time.Second):
t.Fatal("the signal context was never released for a second signal to act on")
}
// Draining, with the game still running: readiness has flipped but the
// player has not been dropped.
deadline := time.Now().Add(drain / 2)
for {
resp, err := http.Get("http://" + addr + "/readyz")
if err == nil {
_ = resp.Body.Close()
if resp.StatusCode == http.StatusServiceUnavailable {
break
}
}
if time.Now().After(deadline) {
t.Fatal("/readyz did not report draining while the game was alive")
}
time.Sleep(20 * time.Millisecond)
}
m := await(func(m *noituv1.ServerMessage) bool { return m.GetError() != nil })
if code := m.GetError().GetCode(); code != "server_restarting" {
t.Errorf("error code = %q, want server_restarting", code)
}
if waited := time.Since(begin); waited < drain-100*time.Millisecond {
t.Errorf("shut down after %v, before the %v drain timeout: the live game was not waited for", waited, drain)
}
select {
case err := <-done:
if err != nil {
t.Errorf("serve returned %v", err)
}
case <-time.After(drain + 10*time.Second):
t.Fatal("serve did not return after the drain timeout")
}
}
func dialWhenUp(t *testing.T, ctx context.Context, addr string) *websocket.Conn {
t.Helper()
var lastErr error
for range 50 {
conn, _, err := websocket.Dial(ctx, "ws://"+addr+"/ws", nil)
if err == nil {
return conn
}
lastErr = err
time.Sleep(50 * time.Millisecond)
}
t.Fatalf("server never came up: %v", lastErr)
return nil
}
// recorder logs the order the shutdown sequence drives the server in, and how
// many games were alive at each step.
type recorder struct {
mu sync.Mutex
games int64
steps []string
}
func (r *recorder) note(step string) {
r.mu.Lock()
defer r.mu.Unlock()
r.steps = append(r.steps, step)
}
func (r *recorder) RoomCount() int { return 1 }
func (r *recorder) LiveGameCount() int64 {
r.mu.Lock()
defer r.mu.Unlock()
return r.games
}
func (r *recorder) StartDraining() {
r.note("drain")
r.mu.Lock()
alive := r.games
r.mu.Unlock()
if alive == 0 {
r.note("drain-with-no-games")
}
}
func (r *recorder) Shutdown() { r.note("shutdown") }
func TestDrainAndShutdownOrdersDrainBeforeShutdown(t *testing.T) {
r := &recorder{games: 1}
start := time.Now()
drainAndShutdown(r, 300*time.Millisecond, 0)
if got := strings.Join(r.steps, ","); got != "drain,shutdown" {
t.Errorf("steps = %q, want drain then shutdown, with games alive at the drain", got)
}
if took := time.Since(start); took < 250*time.Millisecond || took > 5*time.Second {
t.Errorf("returned after %v, want about the 300ms drain timeout", took)
}
}
func TestServersSetOnlyAnIdleTimeout(t *testing.T) {
for name, srv := range map[string]*http.Server{
"public": newHTTPServer(":0", http.NotFoundHandler()),
"debug": newDebugServer(":0"),
} {
if srv.IdleTimeout != idleTimeout {
t.Errorf("%s: IdleTimeout = %v, want %v", name, srv.IdleTimeout, idleTimeout)
}
// A read or write deadline outlives the hijack and would end every
// WebSocket game when it fires.
if srv.ReadTimeout != 0 || srv.WriteTimeout != 0 {
t.Errorf("%s: ReadTimeout/WriteTimeout = %v/%v, want both unset", name, srv.ReadTimeout, srv.WriteTimeout)
}
}
}
func TestCheckHealth(t *testing.T) {
ok := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/healthz" {
http.NotFound(w, r)
}
}))
defer ok.Close()
okAddr := strings.TrimPrefix(ok.URL, "http://")
_, port, _ := net.SplitHostPort(okAddr)
broken := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
http.Error(w, "down", http.StatusServiceUnavailable)
}))
defer broken.Close()
dead, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
deadAddr := dead.Addr().String()
_ = dead.Close()
tests := []struct {
name string
addr string
wantErr bool
}{
{"healthy server", okAddr, false},
{"bare port, as in the default :8080", ":" + port, false},
{"wildcard host is dialled on loopback", "0.0.0.0:" + port, false},
{"non-200 fails", strings.TrimPrefix(broken.URL, "http://"), true},
{"nothing listening fails", deadAddr, true},
{"malformed address fails", "not-an-address", true},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
err := checkHealth(tc.addr, time.Second)
if (err != nil) != tc.wantErr {
t.Errorf("checkHealth(%q) = %v, wantErr %v", tc.addr, err, tc.wantErr)
}
})
}
}