yorickr/plezy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: clean up code comments

Browse Source
This commit is contained in:
edde746
2026-08-10 20:28:41 +02:00
parent 5611c6785a
commit 69fadc220d
170 changed files with 324 additions and 1765 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/artifact_store.go
+13 -28
View File
@@ -14,14 +14,6 @@ import (
"time"
)
// --- Artifact store ---
//
// Uploaded logs and posters are the same on-disk artifact store: a flat
// directory of `<id><ext>` files whose mtime carries the creation time, written
// through a temp file, capped by a quota and swept for expiry. Files that
// cannot be deleted become pending debt so a failed removal never silently
// frees quota. artifactStore implements all of that once; the two flavours
// differ only in the policy fields below.
type artifactRemovalError struct {
err error
@@ -117,8 +109,8 @@ func validID(id string, length int) bool {
return true
}
// pendingRemoval is an artifact file that could not be deleted yet. Its size is
// unknown when the file could not be stat'ed or is not a regular file.
// pendingRemoval tracks an artifact that could not be deleted. Size is unknown
// when stat fails or the path is not a regular file.
type pendingRemoval struct {
size int64
sizeKnown bool
@@ -140,27 +132,22 @@ type artifactStore struct {
maxAge time.Duration
removeFile func(string) error
// Policy. generateID is a field so tests can force ID collisions.
// Policy hooks and quota behavior.
generateID func() string
idFromFilename func(filename string) (string, bool)
// acceptLoaded reports whether a file found on disk is a usable artifact
// and returns the content type recorded for it.
// acceptLoaded returns the content type for a usable on-disk artifact.
acceptLoaded func(filename string, size int64) (string, bool)
// limit caps accountedLocked, measured in the units cost returns:
// one per artifact for logs, bytes for posters.
// limit uses cost units: one per log, bytes per poster.
limit int64
cost func(size int64) int64
pendingCost func(pending pendingRemoval) int64
// evictToFit admits a new artifact by evicting the oldest live ones;
// stores that leave it false reject the upload with errFull instead.
// evictToFit evicts oldest entries instead of rejecting when full.
evictToFit bool
// retryKnownDebtOnPut retries only debt whose size is accounted, leaving
// unknown debt to periodic cleanup.
// retryKnownDebtOnPut retries only size-accounted pending removals.
retryKnownDebtOnPut bool
errFull error
// Pending removals whose cost is not accounted (pendingCost returns 0) are
// tracked only by their pendingRemovals entry; no separate counter exists.
// Unaccounted pending removals are tracked only in pendingRemovals.
used int64 // accounted cost of live entries
pendingDebt int64 // accounted cost of pending removals
startupErr error
@@ -170,11 +157,11 @@ type artifactStore struct {
func (as *artifactStore) filePath(filename string) string {
return filepath.Join(as.dir, filename)
}
func (as *artifactStore) accountedLocked() int64 {
return as.used + as.pendingDebt
}
func (as *artifactStore) loadExisting(now time.Time) error {
as.mu.Lock()
defer as.mu.Unlock()
@@ -230,7 +217,7 @@ func (as *artifactStore) loadExisting(now time.Time) error {
return removalErr
}
// put writes data as `<id><ext>` once the quota allows it.
// put writes data to a quota-approved `<id><ext>` file.
func (as *artifactStore) put(data []byte, ext, contentType string, now time.Time) (string, artifactEntry, error) {
as.mu.Lock()
defer as.mu.Unlock()
@@ -287,9 +274,7 @@ func (as *artifactStore) put(data []byte, ext, contentType string, now time.Time
return id, entry, nil
}
// lookupEntry returns the live entry for id, dropping it when it has expired.
// match, when set, rejects entries the caller did not ask for before expiry is
// considered, so a mismatched request never triggers a removal.
// lookupEntry returns a live matching entry, deleting it if expired.
func (as *artifactStore) lookupEntry(
id string,
now time.Time,
@@ -363,8 +348,8 @@ func (as *artifactStore) deleteEntryLocked(id string) error {
return nil
}
// removeUntrackedLocked deletes a file the index does not own, recording it as
// pending debt when the removal fails.
// removeUntrackedLocked deletes an unindexed file and records failed removal
// as pending debt.
func (as *artifactStore) removeUntrackedLocked(filename string, size int64, sizeKnown bool) error {
if err := removeArtifact(as.removeFile, as.dir, as.filePath(filename)); err != nil {
as.addPendingLocked(filename, size, sizeKnown)
server/main.go
+35 -72
View File
@@ -39,7 +39,7 @@ const (
httpResponseWriteTimeout = oauthResultWait + httpResponseWriteMargin
pongWait = 60 * time.Second
pingInterval = 30 * time.Second
maxLogSize = 1 * 1024 * 1024 // 1MB
maxLogSize = 1 * 1024 * 1024
logMaxAge = 3 * 24 * time.Hour
logIDLength = 5
logRateInterval = 1 * time.Minute
@@ -49,7 +49,7 @@ const (
maxFailedLogLookupSources = 4096
maxConcurrentLogLookups = 32
maxHTTPHeaderBytes = 64 * 1024
maxPosterSize = 5 * 1024 * 1024 // 5MB
maxPosterSize = 5 * 1024 * 1024
maxPosterStoreSize = int64(1 * 1024 * 1024 * 1024)
posterMaxAge = 3 * time.Hour
posterIDLength = 16
@@ -78,7 +78,6 @@ var upgrader = websocket.Upgrader{
CheckOrigin: func(r *http.Request) bool { return true },
}
// --- Messages ---
type clientMsg struct {
Type string `json:"type"`
@@ -104,7 +103,6 @@ type serverMsg struct {
Payload json.RawMessage `json:"payload,omitempty"`
}
// --- Client (serializes writes to a single goroutine) ---
type outboundFrame struct {
data []byte
@@ -216,7 +214,6 @@ func (c *Client) close() {
})
}
// --- Room ---
type reconnectVerifier [sha256.Size]byte
@@ -241,11 +238,8 @@ type Room struct {
LastActivityAt time.Time
}
// --- Snapshot types (on-disk JSON format) ---
// Nanosecond timestamps preserve exact absence state without the expansion of
// RFC3339 strings at the maximum admitted reservation count. Zero means the
// peer was connected when the snapshot was captured.
// Nanosecond timestamps preserve exact absence state; zero means connected.
type peerReservationSnapshot struct {
Verifier string `json:"verifier"`
AbsentSinceUnixNano int64 `json:"absentSince,omitempty"`
@@ -305,8 +299,8 @@ func reconnectVerifierMatches(expected, presented reconnectVerifier) bool {
return subtle.ConstantTimeCompare(expected[:], presented[:]) == 1
}
// pruneExpiredPeerReservationsLocked removes only expired, disconnected guest
// reservations. The caller must hold room.mu.
// pruneExpiredPeerReservationsLocked removes expired, disconnected guest
// reservations; the caller must hold room.mu.
func pruneExpiredPeerReservationsLocked(room *Room, now time.Time) bool {
changed := false
for peerID, reservation := range room.peerReservations {
@@ -417,14 +411,12 @@ func (r *Room) sendFrom(senderID string, sender *Client, targetID string, msg se
return directedTargetFound
}
// --- Log store ---
const logFileExt = ".log"
var errLogStoreFull = errors.New("log store full")
// logStore keeps diagnostic uploads capped by artifact count; a full store
// rejects new uploads rather than evicting logs someone may still be reading.
// logStore rejects uploads when its artifact-count quota is full.
type logStore struct {
artifactStore
rateLimit map[string]time.Time // IP -> last upload time
@@ -521,12 +513,10 @@ func (ls *logStore) cleanup(now time.Time) error {
return removalErr
}
// --- Poster store ---
var errPosterStoreFull = errors.New("poster store full")
// posterStore caps shared posters by accounted bytes and evicts the oldest to
// admit a new upload.
// posterStore evicts oldest artifacts to stay within its byte quota.
type posterStore struct {
artifactStore
}
@@ -649,7 +639,6 @@ func (ps *posterStore) lookup(filename string, now time.Time) (artifactEntry, bo
})
}
// --- Snapshotter (single-writer, debounced, atomic disk persistence) ---
var errSnapshotterStopped = errors.New("snapshot writer is stopped")
@@ -726,9 +715,8 @@ func newSnapshotter(path string, build func() stateSnapshot) *snapshotter {
return sn
}
// recordMutation publishes a protected identity, membership, or reservation
// mutation to the single writer. Callers record after changing state and before
// releasing the lock that made the mutation visible.
// recordMutation publishes a mutation after the caller changes state and before
// releasing the lock that made it visible.
func (sn *snapshotter) recordMutation() uint64 {
sn.stateMu.Lock()
if sn.stopped {
@@ -742,9 +730,7 @@ func (sn *snapshotter) recordMutation() uint64 {
return seq
}
// recordTerminalMutation atomically publishes a protected mutation together
// with its outcome channel. A buffered result retains even an immediate write
// failure until the handler begins waiting.
// recordTerminalMutation publishes a mutation and its buffered outcome channel.
func (sn *snapshotter) recordTerminalMutation(
complete func(error) terminalMutationOutcome,
) *terminalMutationTicket {
@@ -755,9 +741,7 @@ func (sn *snapshotter) recordTerminalMutation(
if complete == nil {
result <- terminalMutationOutcome{err: errSnapshotterStopped, deliver: true}
} else {
// The caller still holds the protected state lock. Run the
// rollback continuation asynchronously so it can acquire the
// normal s.mu -> room.mu order after the caller unlocks.
// Run rollback asynchronously after the caller releases its state lock.
go func() {
result <- complete(errSnapshotterStopped)
}()
@@ -848,8 +832,7 @@ func (sn *snapshotter) run() {
default:
}
}
// Drain tokens queued before capture. A mutation recorded after
// capture re-arms the channels and therefore requires a later write.
// Drain pre-capture tokens; later mutations re-arm the channels.
select {
case <-sn.trigger:
default:
@@ -870,9 +853,8 @@ func (sn *snapshotter) run() {
}
}
// write is the narrowly serialized storage entry retained for atomic-storage
// tests. Production mutations use writeNextGeneration so generation outcomes
// cannot bypass the single writer.
// write is retained for synchronous storage tests; production uses the single
// writer's writeNextGeneration.
func (sn *snapshotter) write() error {
sn.writeMu.Lock()
defer sn.writeMu.Unlock()
@@ -922,9 +904,7 @@ func (sn *snapshotter) writeNextGeneration() (bool, error) {
}
sn.stateMu.Unlock()
// Continuations are part of the writer barrier. In particular, a failed
// staged release rolls back and records its corrective generation before
// this writer can capture any queued later mutation.
// Continuations roll back failed releases before later mutations are captured.
for _, terminal := range covered {
outcome := terminalMutationOutcome{err: err, deliver: true}
if terminal.complete != nil {
@@ -1004,9 +984,8 @@ func (sn *snapshotter) persistAtomic(data []byte) error {
os.Remove(tmpPath)
return err
}
// Rename is the commit boundary: the replacement is file-synced and
// non-torn. Parent-directory sync adds crash durability where supported,
// but its post-commit failure must not report the mutation as uncommitted.
// Rename commits the file-synced replacement. Directory-sync failure is
// warning-only after that boundary.
if err := sn.syncDir(sn.dir); err != nil {
sn.logDirSyncErr(err)
}
@@ -1038,7 +1017,7 @@ func (sn *snapshotter) flushAndStop(timeout time.Duration) error {
return sn.stopErr
}
// logWriteErr throttles pre-commit snapshot-write error spam to once per hour.
// logWriteErr throttles pre-commit snapshot errors to once per hour.
func (sn *snapshotter) logWriteErr(err error) {
sn.errMu.Lock()
defer sn.errMu.Unlock()
@@ -1049,8 +1028,7 @@ func (sn *snapshotter) logWriteErr(err error) {
log.Printf("snapshot: write failed before rename commit: %v", err)
}
// logDirSyncErr has an independent throttle so a degraded post-rename warning
// cannot suppress a later pre-commit persistence error.
// logDirSyncErr independently throttles post-rename directory-sync warnings.
func (sn *snapshotter) logDirSyncErr(err error) {
sn.dirErrMu.Lock()
defer sn.dirErrMu.Unlock()
@@ -1061,7 +1039,6 @@ func (sn *snapshotter) logDirSyncErr(err error) {
log.Printf("snapshot: parent directory sync failed after rename commit: %v", err)
}
// --- Server ---
type removalErrorThrottle struct {
mu sync.Mutex
lastLog map[string]time.Time
@@ -1157,9 +1134,8 @@ func newServer(logDir, stateFile, posterDir string, clientIPs clientIPResolver)
return s
}
// removeRoomLocked removes room only while it is still the authoritative map
// entry. The caller must hold s.mu. A current-process quota reservation follows
// the retained room and is returned exactly once by successful removal.
// removeRoomLocked removes only the authoritative entry and releases its
// current-process quota reservation once.
func (s *Server) removeRoomLocked(sessionID string, room *Room) bool {
if s.rooms[sessionID] != room {
return false
@@ -1171,18 +1147,14 @@ func (s *Server) removeRoomLocked(sessionID string, room *Room) bool {
return true
}
// buildSnapshot is the synchronous storage-test entry. Production capture uses
// captureSnapshot so the copied state and its covered generation share one
// ordering boundary.
// buildSnapshot is the synchronous test entry; production uses captureSnapshot.
func (s *Server) buildSnapshot() stateSnapshot {
snapshot, _ := s.captureSnapshot(func() uint64 { return 0 })
return snapshot
}
// captureSnapshot freezes every durable room mutation under the established
// s.mu -> room.mu order, then captures the covered sequence while those locks
// remain held. A mutation is therefore either both present and covered, or
// neither present nor covered. Locks are released before marshal or disk I/O.
// captureSnapshot holds s.mu -> room.mu while copying state and its covered
// generation, then releases locks before marshal or I/O.
func (s *Server) captureSnapshot(captureSequence func() uint64) (stateSnapshot, uint64) {
s.mu.RLock()
rooms := make([]*Room, 0, len(s.rooms))
@@ -1235,9 +1207,8 @@ func (s *Server) captureSnapshot(captureSequence func() uint64) (stateSnapshot,
return snapshot, targetSeq
}
// loadSnapshot restores rooms from disk on startup. The returned rewrite flag
// reports reservation migration, initialization, or pruning that must be
// persisted before serving. Missing/corrupt files still allow startup.
// loadSnapshot restores rooms. Missing or corrupt files allow startup; the
// rewrite flag requests persistence of migration or pruning.
func (s *Server) loadSnapshot(path string) (bool, error) {
data, err := os.ReadFile(path)
if err != nil {
@@ -1699,7 +1670,7 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) {
http.Error(w, "Invalid client address", http.StatusBadRequest)
return
}
// Retained-room ownership uses the same canonical source key as connection admission.
// Retained-room ownership uses the admission source key.
quotaOwnerKey := ip
if !s.conns.tryConnect(ip) {
@@ -1740,9 +1711,7 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) {
return true
}
// Cleanup on disconnect only when this client is still authoritative. A
// displaced client's defer must neither remove the replacement nor start
// its reservation's absence clock.
// Only the authoritative client may remove the room or start its absence clock.
defer func() {
if currentRoom != nil && currentPeerID != "" {
currentRoom.mu.Lock()
@@ -1883,10 +1852,8 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) {
}
continue
}
// A room nobody is connected to is an abandoned code, not property.
// Whoever asks for it next takes it, so a host that restarted with a
// fresh reconnect token can reuse its own code instead of waiting out
// the cleanup sweep. An occupied room still belongs to its peers.
// An empty room code is abandoned and may be reclaimed; occupied rooms
// remain owned by their peers.
reclaimable := len(existing.Peers) == 0 && !existing.closing
existing.mu.Unlock()
if !reclaimable {
@@ -2038,15 +2005,14 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) {
!occupied &&
room.quotaOwnerKey != "" &&
room.quotaOwnerKey == quotaOwnerKey:
// Tokenless host reconnect is retained only for unversioned rooms,
// only within this process, and only from the creating source.
// Tokenless host reconnect is limited to unversioned local rooms
// from the creating source.
authorized = true
responseToken = ""
responseVerifier = room.hostVerifier
}
} else {
// Legacy guests have no durable proof. Never let one replace a live
// identity; disconnected identity reuse remains confined to legacy rooms.
// Legacy guests lack durable proof, so identity reuse is legacy-only.
authorized = !occupied
}
@@ -2189,8 +2155,7 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) {
if room.LastActivityAt.Equal(leaveActivity) {
room.LastActivityAt = previousActivity
}
// The failed attempt captured the pending omission. Record
// the restored reservation before a queued later capture.
// Record the restored reservation before a queued later capture.
s.snap.recordMutation()
}
room.mu.Unlock()
@@ -2258,9 +2223,7 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) {
room.mu.Unlock()
s.mu.Unlock()
// A successful outcome means the file-synced atomic rename committed.
// Supported filesystems also complete parent-directory sync before
// this barrier; post-rename sync degradation is warning-only.
// Atomic rename committed; post-rename directory-sync degradation is warning-only.
outcome := s.snap.waitForDurable(ticket)
if outcome.err != nil {
client.sendJSON(serverMsg{Type: relayTypeError, Code: relayErrorInvalidMessage, Message: "Unable to persist ended room"})
server/main_test.go
+27 -87
View File
@@ -49,9 +49,7 @@ func TestGeneratedRelayProtocolVersionsMatchSpec(t *testing.T) {
}
}
// newTestServer builds a Server wired for tests: no goroutines and no network.
// Its snapshotter is not started; storage tests drive the narrow synchronous
// write entry directly.
// newTestServer builds a goroutine-free, network-free test server.
func newTestServer(t *testing.T, stateFile string) *Server {
t.Helper()
s := &Server{
@@ -135,7 +133,7 @@ func TestSnapshotRoundTrip(t *testing.T) {
t.Fatalf("snapshot persisted process-local client identity: %s", data)
}
// Reconstruct into a fresh Server and verify identity.
// Reload from disk into a fresh server.
s2 := newTestServer(t, path)
if _, err := s2.loadSnapshot(path); err != nil {
t.Fatalf("loadSnapshot: %v", err)
@@ -229,7 +227,7 @@ func TestLoadHandlesCorrupt(t *testing.T) {
if len(s.rooms) != 0 {
t.Fatalf("expected empty rooms after corrupt load, got %d", len(s.rooms))
}
// File should be preserved for debugging.
// Corrupt snapshots remain available for diagnosis.
if _, err := os.Stat(path); err != nil {
t.Fatalf("corrupt file should NOT be deleted: %v", err)
}
@@ -270,7 +268,7 @@ func TestCleanupUsesIdleNotAge(t *testing.T) {
s := newTestServer(t, filepath.Join(t.TempDir(), "rooms.json"))
now := time.Now()
// 2h-old room that has activity 1min ago — must NOT be cleaned up.
// Recent activity keeps this old room.
s.rooms["KEEP"] = &Room{
SessionID: "KEEP",
HostPeerID: "h",
@@ -278,7 +276,7 @@ func TestCleanupUsesIdleNotAge(t *testing.T) {
CreatedAt: now.Add(-2 * time.Hour),
LastActivityAt: now.Add(-1 * time.Minute),
}
// 2h-old room that emptied 10min ago — MUST be cleaned up.
// Idle rooms are removed.
s.rooms["GONE"] = &Room{
SessionID: "GONE",
HostPeerID: "h",
@@ -286,11 +284,11 @@ func TestCleanupUsesIdleNotAge(t *testing.T) {
CreatedAt: now.Add(-2 * time.Hour),
LastActivityAt: now.Add(-10 * time.Minute),
}
// 25h-old room — absolute TTL nukes it even if recently active.
// The absolute TTL removes this room despite recent activity.
s.rooms["OLD"] = &Room{
SessionID: "OLD",
HostPeerID: "h",
Peers: map[string]*Client{}, // empty anyway
Peers: map[string]*Client{},
CreatedAt: now.Add(-25 * time.Hour),
LastActivityAt: now.Add(-10 * time.Second),
}
@@ -352,7 +350,7 @@ func TestSnapshotAtomicWriteSurvivesRenameFailure(t *testing.T) {
path := filepath.Join(dir, "rooms.json")
s := newTestServer(t, path)
// Seed a valid snapshot on disk.
// Seed the on-disk snapshot.
s.rooms["ORIG"] = &Room{
SessionID: "ORIG",
HostPeerID: "h",
@@ -368,8 +366,7 @@ func TestSnapshotAtomicWriteSurvivesRenameFailure(t *testing.T) {
t.Fatalf("read orig: %v", err)
}
// Block the temporary-file open with a directory at the same path. This
// deterministically fails before rename on every supported platform.
// A directory at the temporary path makes the open fail before rename.
if err := os.Mkdir(path+".tmp", 0755); err != nil {
t.Fatalf("create blocking temporary directory: %v", err)
}
@@ -598,7 +595,7 @@ func TestSnapshotDebounceCoalesces(t *testing.T) {
go sn.run()
t.Cleanup(func() { _ = sn.flushAndStop(time.Second) })
// Fire a burst — should collapse into one write due to debounce.
// A burst should collapse into one debounced write.
for i := 0; i < 20; i++ {
sn.recordMutation()
}
@@ -1132,11 +1129,6 @@ func TestSnapshotDirectorySyncWarningIsThrottled(t *testing.T) {
}
}
// ======================================================================
// Integration harness — boots a real Server behind httptest with the full
// HTTP mux. Each dial sets X-Forwarded-For so tests control the perceived
// client IP independently of the rate limiters.
// ======================================================================
type relayHarness struct {
srv *Server
@@ -1171,8 +1163,7 @@ func newRelayHarnessNoTrust(t *testing.T) *relayHarness {
)
}
// newRelayHarnessAt lets a test control the stateFile path so two harnesses
// can share a snapshot across a simulated restart.
// newRelayHarnessAt allows two harnesses to share a snapshot across restarts.
func newRelayHarnessAt(t *testing.T, logDir, stateFile string) *relayHarness {
t.Helper()
return newRelayHarnessAtWithResolver(t, logDir, stateFile, mustClientIPResolver(t, "127.0.0.0/8"))
@@ -1501,8 +1492,7 @@ func (c *testConn) expectAuthority(typ, hostPeerID string) serverMsg {
return message
}
// recvNothing asserts no message arrives within the given window. Used to
// verify silent paths (sender not receiving own broadcast, stale-peer skip).
// recvNothing asserts that no frame arrives within the window.
func (c *testConn) recvNothing(within time.Duration) {
c.t.Helper()
c.conn.SetReadDeadline(time.Now().Add(within))
@@ -1515,8 +1505,7 @@ func (c *testConn) recvNothing(within time.Duration) {
}
}
// recvUntilClosed consumes any frames already queued on the wire and requires
// a permanent terminal read error before the absolute deadline.
// recvUntilClosed drains queued frames and waits for terminal closure.
func (c *testConn) recvUntilClosed(within time.Duration) ([]serverMsg, error) {
c.t.Helper()
if err := c.conn.SetReadDeadline(time.Now().Add(within)); err != nil {
@@ -1681,9 +1670,6 @@ func TestClientWriteFailureClosesConnection(t *testing.T) {
client.close()
}
// ======================================================================
// Unit tests — pure logic
// ======================================================================
func TestRateLimiterBurstExhausts(t *testing.T) {
rl := newRateLimiter(5, 10)
@@ -1698,7 +1684,7 @@ func TestRateLimiterBurstExhausts(t *testing.T) {
}
func TestRateLimiterRefillsOverTime(t *testing.T) {
rl := newRateLimiter(5, 10) // 10 tokens/sec
rl := newRateLimiter(5, 10)
for i := 0; i < 5; i++ {
rl.allow()
}
@@ -1734,8 +1720,7 @@ func TestRateLimiterAllowRace(t *testing.T) {
}()
}
wg.Wait()
// Real assertion is that -race finds no data race. Spot-check the
// result is within plausible bounds.
// Spot-check the result bounds; -race checks synchronization.
if got := successes.Load(); got <= 0 || got > 500 {
t.Fatalf("unexpected successes count %d (want 1..500)", got)
}
@@ -1775,9 +1760,6 @@ func TestCleanupRateWindowsUsesWindowBoundary(t *testing.T) {
}
}
// ======================================================================
// connTracker unit tests
// ======================================================================
func TestConnTrackerPerIPLimit(t *testing.T) {
ct := newConnTracker()
@@ -1821,7 +1803,7 @@ func TestConnTrackerDisconnectFrees(t *testing.T) {
t.Errorf("globalCount=%d, want 0", ct.globalCount)
}
ct.mu.Unlock()
// Extra disconnect is a no-op (doesn't panic).
// Extra disconnect is a no-op.
ct.disconnect(ip)
}
@@ -1881,9 +1863,7 @@ func TestConnTrackerConnectRateLimit(t *testing.T) {
t.Fatalf("warmup tryConnect %d: expected true", i)
}
}
// Free one slot so the perIP check won't be what rejects us.
ct.disconnect(ip)
// Rate-limit bucket is empty now; this should be the denial path.
// Free a slot so the next denial comes from the rate limiter.
if ct.tryConnect(ip) {
t.Fatal("expected false from rate-limit bucket, not per-IP cap")
}
@@ -1990,9 +1970,6 @@ func TestPosterUploadLimiterAdmissionPolicy(t *testing.T) {
})
}
// ======================================================================
// clientIPResolver unit tests
// ======================================================================
func TestClientIPResolverTrustChains(t *testing.T) {
tests := []struct {
@@ -2088,12 +2065,8 @@ func TestParseTrustedProxyCIDRs(t *testing.T) {
}
}
// ======================================================================
// generateLogID
// ======================================================================
// Random ids may legitimately repeat, so shape is the only contract here;
// collision retry is covered deterministically by
// Random IDs may repeat; this test checks shape. Collision retry is covered by
// TestLogStorePersistsAcrossRestartAndAvoidsIDCollisions.
func TestGenerateLogIDShape(t *testing.T) {
for range 200 {
@@ -2109,9 +2082,6 @@ func TestGenerateLogIDShape(t *testing.T) {
}
}
// ======================================================================
// handleWS — create case
// ======================================================================
func TestCreateSucceeds(t *testing.T) {
h := newRelayHarness(t)
@@ -2144,7 +2114,7 @@ func TestCreateDuplicateReturnsRoomExists(t *testing.T) {
c1.send(clientMsg{Type: "create", SessionID: "SAME", PeerID: "host-1"})
c1.expect("created")
// Different IP to avoid the per-IP rooms quota interfering.
// Use a different IP so the rooms quota does not interfere.
c2 := h.dial(t, "1.1.1.5")
c2.send(clientMsg{Type: "create", SessionID: "SAME", PeerID: "host-2"})
c2.expectError("room_exists")
@@ -2308,8 +2278,7 @@ func TestCreateReclaimsAbandonedEmptyRoom(t *testing.T) {
t.Fatal("abandoned room identity survived the reclaim")
}
// The previous owner's capability died with the room it belonged to, and
// the live replacement is not reclaimable by anyone, owner included.
// The former capability cannot reclaim the live replacement.
former := h.dial(t, "1.1.1.60")
former.send(clientMsg{
Type: relayTypeCreate,
@@ -2321,9 +2290,7 @@ func TestCreateReclaimsAbandonedEmptyRoom(t *testing.T) {
former.expectError(relayErrorRoomExists)
}
// The recent-rooms flow: a host restarts its app, so it presents a fresh
// reconnect capability for a code the relay still holds. The abandoned code
// must come back as a hosted room instead of a ghost room with no host.
// A restarted host presents a fresh capability for an abandoned room code.
func TestAbandonedCodeIsRecreatableByARestartedHost(t *testing.T) {
h := newRelayHarness(t)
firstToken, _ := mustReconnectToken(t)
@@ -2339,8 +2306,7 @@ func TestAbandonedCodeIsRecreatableByARestartedHost(t *testing.T) {
host.conn.Close()
h.waitRoomPeers(t, "REUSE", 0)
// A restarted app mints a new capability, so it cannot prove the previous
// ownership even when it reuses its own peer ID.
// A fresh capability cannot prove previous ownership, even with the same peer ID.
restartToken, _ := mustReconnectToken(t)
restarted := h.dial(t, "6.4.0.2")
restarted.send(clientMsg{
@@ -2423,7 +2389,7 @@ func TestCreateHitsRoomsPerIPLimit(t *testing.T) {
c.send(clientMsg{Type: "create", SessionID: fmt.Sprintf("R%d", i), PeerID: "host"})
c.expect("created")
}
// 4th create from same IP exceeds the quota.
// The fourth room from this IP exceeds the quota.
c := h.dial(t, ip)
c.send(clientMsg{Type: "create", SessionID: "ROVERFLOW", PeerID: "host"})
c.expectError("rate_limited")
@@ -2782,9 +2748,6 @@ func TestConnectionCannotRetainMultipleRoomMemberships(t *testing.T) {
}
}
// ======================================================================
// handleWS — join case
// ======================================================================
func TestJoinSucceedsAndBroadcastsPeerJoined(t *testing.T) {
h := newRelayHarness(t)
@@ -3345,9 +3308,6 @@ func TestJoinAdmissionIsAtomicWithReservedRoomCreate(t *testing.T) {
}
}
// ======================================================================
// handleWS — broadcast / sendTo
// ======================================================================
func TestBroadcastDeliversToOthersNotSender(t *testing.T) {
h := newRelayHarness(t)
@@ -3381,7 +3341,7 @@ func TestBroadcastDeliversToOthersNotSender(t *testing.T) {
t.Errorf("g2 From=%q want G1", g2Msg.From)
}
// Sender should not receive its own broadcast.
// Broadcasts exclude the sender.
g1.recvNothing(200 * time.Millisecond)
}
@@ -3490,9 +3450,6 @@ func TestSendToNotInRoomRejected(t *testing.T) {
c.expectError("not_in_room")
}
// ======================================================================
// handleWS — ping / misc / rate limits
// ======================================================================
func TestPingReturnsPong(t *testing.T) {
h := newRelayHarness(t)
@@ -3521,7 +3478,7 @@ func TestPerConnectionMessageRateLimit(t *testing.T) {
c.send(clientMsg{Type: "create", SessionID: "RL", PeerID: "H"})
c.expect("created")
// The per-connection bucket is rateBurst=30. After ~30 pings we start seeing rate_limited.
// Exceed the per-connection bucket and observe rate limiting.
sawRateLimit := false
for i := 0; i < rateBurst+10; i++ {
c.send(clientMsg{Type: "ping"})
@@ -3538,9 +3495,6 @@ func TestPerConnectionMessageRateLimit(t *testing.T) {
}
}
// ======================================================================
// handleWS — disconnect lifecycle
// ======================================================================
func TestDisconnectBroadcastsPeerLeft(t *testing.T) {
h := newRelayHarness(t)
@@ -5364,9 +5318,7 @@ func TestHostEndDeliversEndedAfterConcurrentGuestTraffic(t *testing.T) {
t.Fatal("ending room remained discoverable before terminal delivery")
}
// WebSocket frames are processed in order. Receiving pong proves the
// preceding membership-sensitive traffic was handled while ended delivery
// was blocked, without closing the guest as a stale client.
// Ordered frames prove membership traffic completed while ended delivery waited.
guest.send(clientMsg{
Type: relayTypeBroadcast,
Payload: json.RawMessage(`{"during":"end"}`),
@@ -5543,9 +5495,6 @@ func TestCleanupDisconnectsPeersBeforeRemovingExpiredOccupiedRoom(t *testing.T)
}
}
// ======================================================================
// Logs endpoints
// ======================================================================
func postLog(t *testing.T, baseURL, ip string, body []byte) *http.Response {
t.Helper()
@@ -5579,8 +5528,6 @@ func getLog(t *testing.T, baseURL, ip, id string) *http.Response {
return resp
}
// postLogAndGetID uploads a log and returns the generated id, asserting the
// POST succeeded.
func postLogAndGetID(t *testing.T, baseURL, ip string, body []byte) string {
t.Helper()
resp := postLog(t, baseURL, ip, body)
@@ -6082,7 +6029,7 @@ func TestLogsUploadDoesNotWriteCapabilityToOperationalLog(t *testing.T) {
func TestLogStoreRetiresLegacyCapabilitiesOnStartup(t *testing.T) {
dir := t.TempDir()
now := time.Now().Add(-time.Minute)
legacyID := strings.Repeat("a", 25) // capability shape used before ids went back to logIDLength
legacyID := strings.Repeat("a", 25) // legacy capability length
currentID := strings.Repeat("a", logIDLength)
legacyPath := filepath.Join(dir, legacyID+".log")
currentPath := filepath.Join(dir, currentID+".log")
@@ -6365,7 +6312,6 @@ func TestLogsGetExpiredIs404(t *testing.T) {
h := newRelayHarness(t)
id := postLogAndGetID(t, h.baseURL, "7.3.0.1", []byte("temp"))
// Poison the entry's ExpiresAt into the past.
h.srv.logs.mu.Lock()
entry := h.srv.logs.entries[id]
entry.ExpiresAt = time.Now().Add(-time.Minute)
@@ -6394,9 +6340,6 @@ func TestLogsMethodNotAllowed(t *testing.T) {
}
}
// ======================================================================
// Poster endpoints
// ======================================================================
var minimalPNG = []byte{0x89, 'P', 'N', 'G', 0x0d, 0x0a, 0x1a, 0x0a, 0x01, 0x02, 0x03}
@@ -7687,9 +7630,6 @@ func TestRemovalFailureLogDoesNotExposeCapabilityPath(t *testing.T) {
}
}
// ======================================================================
// End-to-end: rooms survive a process restart
// ======================================================================
func TestSnapshotSurvivesRestartWithHostAuthority(t *testing.T) {
stateFile := filepath.Join(t.TempDir(), "rooms.json")
server/oauth.go
+25 -42
View File
@@ -1,10 +1,7 @@
package main
// OAuth proxy: relays MAL + AniList authorization-code flows for devices that
// can't listen on localhost (TVs) or lack a browser (headless set-top boxes
// pair via a phone QR scan). Sessions live in memory for 10 minutes; AniList's
// client secret lives only in env vars. Access tokens transit the server
// briefly during code→token exchange and are never logged or persisted.
// OAuth proxy for TV/headless clients. Sessions remain in memory for 10 minutes;
// client secrets are environment-only and tokens are never logged or persisted.
import (
"context"
@@ -31,14 +28,13 @@ const (
oauthMaxSessions = 5000
oauthStartBurst = 3
oauthStartRateSustained = 1
oauthBrowserStateBytes = 18 // 144 bits → 24 base64url chars
oauthPollSecretBytes = 18 // Independently generated device capability.
oauthBrowserStateBytes = 18 // 144 bits, 24 base64url characters
oauthPollSecretBytes = 18 // independent device capability
oauthPKCEVerifierLen = 64
oauthUpstreamTimeout = 15 * time.Second
)
// oauthServiceConfig describes a single upstream OAuth provider. Populated from
// env vars in oauthConfigFromEnv. A service with an empty ClientID is disabled.
// oauthServiceConfig describes one configured upstream provider.
type oauthServiceConfig struct {
ClientID string
ClientSecret string // empty ⇒ provider doesn't issue/require one (MAL w/ PKCE)
@@ -56,10 +52,8 @@ type oauthTokenResult struct {
Error string `json:"error,omitempty"`
}
// oauthSession is created by /auth/start and lives until its result is claimed
// or it is removed by cleanup. browserState crosses the browser/provider trust
// boundary. Only the SHA-256 digest of the device-only poll secret is retained.
// When both locks are needed, oauthProxy.mu must be acquired before s.mu.
// oauthSession links browser state to a device-only poll capability. Only the
// poll secret digest is retained; acquire oauthProxy.mu before s.mu.
type oauthSession struct {
browserState string
pollDigest [sha256.Size]byte
@@ -72,7 +66,7 @@ type oauthSession struct {
completed bool
result *oauthTokenResult
// Test seam used to deterministically seat concurrent result waiters.
// Test seam for deterministic concurrent waiter tests.
waitStarted func()
}
@@ -88,8 +82,8 @@ func (s *oauthSession) completeLocked(r oauthTokenResult) bool {
return true
}
// wait blocks only until the session is ready or ctx is cancelled. Result
// ownership is transferred separately by oauthProxy.claimResult.
// wait blocks until the session is ready or ctx is cancelled. Result ownership
// transfers separately through oauthProxy.claimResult.
func (s *oauthSession) wait(ctx context.Context) error {
if s.waitStarted != nil {
s.waitStarted()
@@ -109,7 +103,7 @@ func (s *oauthSession) pkceVerifier() string {
}
type oauthProxy struct {
baseURL string // e.g. https://ice.plezy.app
baseURL string
services map[string]oauthServiceConfig
client *http.Client
clientIPs clientIPResolver
@@ -134,9 +128,7 @@ func newOAuthProxy(baseURL string, services map[string]oauthServiceConfig, clien
}
}
// oauthConfigFromEnv reads the public base URL and per-service creds from the
// environment. Returns (nil, false) if OAUTH_BASE_URL is unset — the caller
// wires this as "OAuth disabled, endpoints return 503".
// oauthConfigFromEnv returns disabled when OAUTH_BASE_URL is unset.
func oauthConfigFromEnv(clientIPs clientIPResolver) (*oauthProxy, bool) {
base := os.Getenv("OAUTH_BASE_URL")
if base == "" {
@@ -163,8 +155,7 @@ func oauthConfigFromEnv(clientIPs clientIPResolver) (*oauthProxy, bool) {
return newOAuthProxy(base, services, clientIPs), true
}
// registerOAuthRoutes registers all /auth/* handlers. If p is nil (no env
// config), all paths 503 so the integration page clearly says "not configured".
// registerOAuthRoutes mounts /auth/*; a nil proxy returns 503.
func registerOAuthRoutes(mux *http.ServeMux, p *oauthProxy) {
if p == nil {
mux.HandleFunc("/auth/", func(w http.ResponseWriter, r *http.Request) {
@@ -178,8 +169,7 @@ func registerOAuthRoutes(mux *http.ServeMux, p *oauthProxy) {
mux.HandleFunc("/auth/", p.handleAuthRoot)
}
// handleAuthRoot dispatches /auth/... paths that aren't served by their own
// registered handler. Covers /auth/:service and /auth/:service/callback.
// handleAuthRoot dispatches /auth/:service and /auth/:service/callback.
func (p *oauthProxy) handleAuthRoot(w http.ResponseWriter, r *http.Request) {
rest := strings.TrimPrefix(r.URL.Path, "/auth/")
parts := strings.SplitN(rest, "/", 2)
@@ -199,8 +189,7 @@ func (p *oauthProxy) handleAuthRoot(w http.ResponseWriter, r *http.Request) {
http.NotFound(w, r)
}
// POST /auth/start body={"service":"mal"|"anilist"}
// Response: {"session":"device-only poll capability","url":"https://.../auth/:service?state=...","expiresIn":600}
// POST /auth/start returns a device poll capability and authorization URL.
func (p *oauthProxy) handleStart(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Cache-Control", "no-store, private")
if r.Method != http.MethodPost {
@@ -230,8 +219,7 @@ func (p *oauthProxy) handleStart(w http.ResponseWriter, r *http.Request) {
return
}
// Generate independent trust-domain values outside the map lock —
// crypto/rand syscalls must not serialize concurrent /auth/start calls.
// Generate trust-domain values outside the map lock; crypto/rand can block.
var pollSecret string
var sess *oauthSession
for {
@@ -270,7 +258,7 @@ func (p *oauthProxy) handleStart(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, resp)
}
// GET /auth/:service?state=X → 302 upstream authorize URL
// GET /auth/:service?state=X redirects to the upstream authorize URL.
func (p *oauthProxy) handleAuthorize(w http.ResponseWriter, r *http.Request, service string) {
if r.Method != http.MethodGet {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
@@ -306,7 +294,7 @@ func (p *oauthProxy) handleAuthorize(w http.ResponseWriter, r *http.Request, ser
http.Redirect(w, r, cfg.AuthorizeURL+"?"+q.Encode(), http.StatusFound)
}
// GET /auth/:service/callback?code=...&state=... → exchange, park, render page
// GET /auth/:service/callback exchanges the code and renders a result page.
func (p *oauthProxy) handleCallback(w http.ResponseWriter, r *http.Request, service string) {
if r.Method != http.MethodGet {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
@@ -368,7 +356,7 @@ func (p *oauthProxy) handleCallback(w http.ResponseWriter, r *http.Request, serv
renderSuccessPage(w)
}
// GET /auth/result?session=X → long-poll, returns one terminal result.
// GET /auth/result?session=X long-polls for one terminal result.
func (p *oauthProxy) handleResult(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Cache-Control", "no-store, private")
if r.Method != http.MethodGet {
@@ -403,7 +391,7 @@ func (p *oauthProxy) handleResult(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, result)
}
// GET /auth/done — static success page (Simkl's redirect target).
// GET /auth/done renders the static OAuth success page.
func (p *oauthProxy) handleDone(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
@@ -464,15 +452,13 @@ func (p *oauthProxy) redirectURI(service string) string {
return fmt.Sprintf("%s/auth/%s/callback", p.baseURL, service)
}
// addSessionLocked installs both independently generated keys as one logical
// session. The caller has already verified that neither key is live.
// addSessionLocked installs both keys for one session after collision checks.
func (p *oauthProxy) addSessionLocked(sess *oauthSession) {
p.browserStates[sess.browserState] = sess
p.pollDigests[sess.pollDigest] = sess
}
// removeSessionLocked removes only entries still owned by sess, so a stale
// callback or waiter cannot remove a replacement.
// removeSessionLocked removes only entries still owned by sess.
func (p *oauthProxy) removeSessionLocked(sess *oauthSession) {
if p.browserStates[sess.browserState] == sess {
delete(p.browserStates, sess.browserState)
@@ -510,7 +496,7 @@ func (p *oauthProxy) claimResult(digest [sha256.Size]byte, sess *oauthSession) (
return result, true
}
// cleanup drops sessions past oauthSessionTTL. Called by the main cleanup loop.
// cleanup drops sessions past oauthSessionTTL.
func (p *oauthProxy) cleanup() {
now := time.Now()
p.mu.Lock()
@@ -539,7 +525,7 @@ func (p *oauthProxy) ipAllow(ip string) bool {
const successPageHTML = `<!doctype html><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1"><title>Signed in</title><style>html,body{margin:0;height:100%}body{display:flex;flex-direction:column;align-items:center;justify-content:center;font-family:-apple-system,system-ui,sans-serif;background:#fff;color:#1a1a1a;text-align:center;padding:1em;box-sizing:border-box}@media(prefers-color-scheme:dark){body{background:#0f0f0f;color:#f5f5f5}}.check{width:72px;height:72px;margin-bottom:20px}h2{margin:0 0 8px;font-weight:600;font-size:1.25rem}p{margin:0;opacity:.7;font-size:.95rem}</style><body><svg class="check" viewBox="0 0 24 24"><circle cx="12" cy="12" r="10" fill="#22c55e"/><path d="M7 12.5l3 3 7-7" stroke="#fff" stroke-width="2" fill="none" stroke-linecap="round" stroke-linejoin="round"/></svg><h2>Signed in to Plezy</h2><p>You can close this tab and return to the app.</p></body>`
// Split around the message so CSS `%` literals don't collide with Fprintf verbs.
// Keep CSS percent literals outside Fprintf's format string.
const errorPagePrefix = `<!doctype html><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1"><title>Sign-in failed</title><style>html,body{margin:0;height:100%}body{display:flex;flex-direction:column;align-items:center;justify-content:center;font-family:-apple-system,system-ui,sans-serif;background:#fff;color:#1a1a1a;text-align:center;padding:1em;box-sizing:border-box}@media(prefers-color-scheme:dark){body{background:#0f0f0f;color:#f5f5f5}}.x{width:72px;height:72px;margin-bottom:20px}h2{margin:0 0 8px;font-weight:600;font-size:1.25rem}p{margin:0;opacity:.7;font-size:.95rem}</style><body><svg class="x" viewBox="0 0 24 24"><circle cx="12" cy="12" r="10" fill="#ef4444"/><path d="M8 8l8 8M16 8l-8 8" stroke="#fff" stroke-width="2" fill="none" stroke-linecap="round"/></svg><h2>Sign-in failed</h2><p>`
const errorPageSuffix = `</p></body>`
@@ -570,15 +556,12 @@ func digestPollSecret(secret string) [sha256.Size]byte {
func randToken(numBytes int) string {
b := make([]byte, numBytes)
if _, err := rand.Read(b); err != nil {
// crypto/rand failing is catastrophic; log.Fatalf matches the style
// in newLogStore for similar unrecoverable init failures.
log.Fatalf("crypto/rand: %v", err)
}
return base64.RawURLEncoding.EncodeToString(b)
}
// randPKCEVerifier returns a 64-char string from MAL's required alphabet
// (RFC 7636 §4.1 unreserved set).
// randPKCEVerifier returns MAL's RFC 7636 §4.1 verifier alphabet.
func randPKCEVerifier() string {
const alphabet = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-._~"
b := make([]byte, oauthPKCEVerifierLen)
server/oauth_test.go
+8 -23
View File
@@ -15,9 +15,7 @@ import (
"time"
)
// mockUpstream runs an httptest server that impersonates MAL/AniList. It
// records the last token-exchange form submission and returns a canned
// access_token/refresh_token response.
// mockUpstream records token exchanges and returns a canned response.
type mockUpstream struct {
srv *httptest.Server
mu sync.Mutex
@@ -26,9 +24,7 @@ type mockUpstream struct {
tokenCode int
}
// httpGet / httpPost / httpDo wrap the stdlib calls to fail the test on error.
// Keeps test bodies one-liner without tripping `go vet`'s
// "using resp before checking errors" rule.
// Request helpers fail tests on client errors.
func httpGet(t *testing.T, url string) *http.Response {
t.Helper()
resp, err := http.Get(url)
@@ -65,7 +61,7 @@ func newMockUpstream(t *testing.T) *mockUpstream {
m.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch r.URL.Path {
case "/oauth/authorize":
// Unused in tests — we assert on the 302 Location from our proxy.
// Authorization is asserted on the proxy's redirect.
w.WriteHeader(http.StatusOK)
case "/oauth/token":
if err := r.ParseForm(); err != nil {
@@ -101,8 +97,7 @@ func (m *mockUpstream) form() url.Values {
return m.lastForm
}
// newOAuthHarness boots a relay-less httptest server that mounts /auth/* only,
// with `mal` and `anilist` services pointed at a shared mock upstream.
// newOAuthHarness mounts /auth/* against shared mock providers.
type oauthHarness struct {
proxy *oauthProxy
srv *httptest.Server
@@ -137,7 +132,7 @@ func newOAuthHarnessWithResolver(t *testing.T, clientIPs clientIPResolver) *oaut
registerOAuthRoutes(mux, proxy)
srv := httptest.NewServer(mux)
t.Cleanup(srv.Close)
// Rewire baseURL to the real httptest URL so redirect_uri computes correctly.
// Use the real server URL when constructing redirect_uri.
proxy.baseURL = srv.URL
return &oauthHarness{proxy: proxy, srv: srv, base: srv.URL, upstream: up}
}
@@ -197,7 +192,6 @@ func postOAuthStart(t *testing.T, h *oauthHarness, service, xff string) *http.Re
return httpDo(t, req)
}
// ====== /auth/start ======
func TestOAuthStartSeparatesDeviceCapabilityFromBrowserState(t *testing.T) {
h := newOAuthHarness(t)
@@ -247,7 +241,7 @@ func TestOAuthStartRateLimitedPerIP(t *testing.T) {
h := newOAuthHarness(t)
ip := "5.5.5.5"
for range oauthStartBurst {
_, _, _ = h.startSession(t, "mal", ip) // should all succeed
_, _, _ = h.startSession(t, "mal", ip)
}
body, _ := json.Marshal(map[string]string{"service": "mal"})
req, err := http.NewRequest(http.MethodPost, h.base+"/auth/start", bytes.NewReader(body))
@@ -364,7 +358,6 @@ func TestOAuthStartMethodNotAllowed(t *testing.T) {
}
}
// ====== /auth/:service (authorize redirect) ======
func TestOAuthAuthorizeMALRedirectIncludesPKCE(t *testing.T) {
h := newOAuthHarness(t)
@@ -439,7 +432,7 @@ func TestOAuthAuthorizeUnknownSessionRendersError(t *testing.T) {
func TestOAuthAuthorizeWrongServiceRejected(t *testing.T) {
h := newOAuthHarness(t)
_, browserState, _ := h.startSession(t, "mal", "1.1.1.3")
// Try to use the MAL browser state against the AniList authorize endpoint.
// A browser state is valid only for its original service.
resp := httpGet(t, h.base+"/auth/anilist?state="+url.QueryEscape(browserState))
defer resp.Body.Close()
if resp.StatusCode != http.StatusNotFound {
@@ -447,7 +440,6 @@ func TestOAuthAuthorizeWrongServiceRejected(t *testing.T) {
}
}
// ====== /auth/:service/callback + /auth/result ======
type oauthResultResponse struct {
status int
@@ -693,7 +685,6 @@ func TestOAuthCallbackUnknownSessionIgnored(t *testing.T) {
}
}
// ====== Cleanup ======
func TestOAuthCleanupRemovesBothIndexesAndSuppressesStaleCompletion(t *testing.T) {
h := newOAuthHarness(t)
@@ -725,7 +716,6 @@ func TestOAuthCleanupRemovesBothIndexesAndSuppressesStaleCompletion(t *testing.T
}
}
// ====== /auth/done ======
func TestOAuthDoneRendersSuccessPage(t *testing.T) {
h := newOAuthHarness(t)
@@ -740,7 +730,6 @@ func TestOAuthDoneRendersSuccessPage(t *testing.T) {
}
}
// ====== Disabled proxy returns 503 ======
func TestOAuthRoutesReturn503WhenDisabled(t *testing.T) {
mux := http.NewServeMux()
@@ -755,7 +744,6 @@ func TestOAuthRoutesReturn503WhenDisabled(t *testing.T) {
}
}
// ====== Path dispatch ======
func TestOAuthAuthRootRejectsBadPaths(t *testing.T) {
h := newOAuthHarness(t)
@@ -768,12 +756,9 @@ func TestOAuthAuthRootRejectsBadPaths(t *testing.T) {
}
}
// ====== Long-poll timeout ======
func TestOAuthResultBlocksUntilCancel(t *testing.T) {
// Pending sessions must NOT respond immediately; the long-poll contract is
// that /auth/result blocks until the session completes or the client
// cancels. The 204-after-server-timeout path takes 50s so isn't asserted.
// Pending sessions must block until completion or client cancellation.
h := newOAuthHarness(t)
sess, _, _ := h.startSession(t, "mal", "5.5.5.1")
server/rate_limit.go
+5 -13
View File
@@ -5,7 +5,6 @@ import (
"time"
)
// --- Rate limiter (token bucket) ---
type rateLimiter struct {
tokens float64
@@ -65,8 +64,7 @@ func (rl *rateLimiter) refillAtLocked(now time.Time) {
}
}
// reclaimable reports whether discarding this limiter would preserve its
// behavior: enough idle time has passed for the bucket to be full again.
// reclaimable is true when the bucket has refilled completely.
func (rl *rateLimiter) reclaimable(now time.Time) bool {
rl.mu.Lock()
defer rl.mu.Unlock()
@@ -90,7 +88,6 @@ func cleanupRateWindows(windows map[string]time.Time, now time.Time, duration ti
}
}
// --- Poster upload admission (global, per-IP, and concurrency) ---
type posterUploadLimiter struct {
mu sync.Mutex
@@ -152,7 +149,6 @@ func (pl *posterUploadLimiter) cleanup(now time.Time) {
cleanupRateLimiters(pl.perIP, now, nil)
}
// --- Connection tracker (per-IP limits) ---
type connTracker struct {
mu sync.Mutex
@@ -186,8 +182,7 @@ func (ct *connTracker) tryConnect(ip string) bool {
rl = newRateLimiter(connRateBurst, connRateSustained)
ct.ipRate[ip] = rl
}
// Unlock ct.mu before calling rl.allow() would be cleaner,
// but since rl has its own mutex this is safe (no deadlock).
// rl has its own mutex, so holding ct.mu here cannot deadlock.
if !rl.allow() {
return false
}
@@ -210,9 +205,7 @@ func (ct *connTracker) disconnect(ip string) {
}
}
// tryCreateRoom reserves capacity for a retained room created in this process.
// The reservation survives creator disconnect and is released only when the
// authoritative room is removed from Server.rooms.
// tryCreateRoom reserves capacity until the retained room is removed.
func (ct *connTracker) tryCreateRoom(ip string) bool {
ct.mu.Lock()
defer ct.mu.Unlock()
@@ -223,9 +216,8 @@ func (ct *connTracker) tryCreateRoom(ip string) bool {
return true
}
// tryCreateRoomReplacing reserves a room while accounting for the reservation
// that removeRoomLocked will immediately release from an empty same-ID room.
// Server.mu serializes this paired reservation/removal transaction.
// tryCreateRoomReplacing accounts for removal of an empty same-ID room.
// Server.mu serializes the reservation/removal transaction.
func (ct *connTracker) tryCreateRoomReplacing(ip, replacedOwnerKey string) bool {
ct.mu.Lock()
defer ct.mu.Unlock()