okooooooo/Piik
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: preserve room authority and repair Browser/Client recovery (#386)

Browse Source 
* fix: preserve room authority and recover current media connections

* docs: close room authority and recovery integration
This commit is contained in:
AAA authored and GitHub committed 2026-09-10 01:30:07 +08:00
1 parent c9a6642ac1
commit 35f9251088
67 files changed
+1553 -1441
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

internal/server/app/app.go
+2 -6
View File
@@ -66,7 +66,6 @@ type Options struct {
AuthenticationTimeoutMs int
ViewerDisconnectGraceMs int
HeartbeatIntervalMs int
CleanupIntervalMs int
MaxSignalConnections int
MaxUnauthenticatedSignalConnections int
SiteAccessTTLSeconds int
@@ -125,7 +124,7 @@ func New(options Options) (*Server, error) {
logger = slog.Default()
}
store, err := newRoomStore(options, now)
store, err := newRoomStore(options)
if err != nil {
return nil, err
}
@@ -171,7 +170,6 @@ func New(options Options) (*Server, error) {
AuthenticationTimeoutMs: options.AuthenticationTimeoutMs,
ViewerDisconnectGraceMs: options.ViewerDisconnectGraceMs,
HeartbeatIntervalMs: options.HeartbeatIntervalMs,
CleanupIntervalMs: options.CleanupIntervalMs,
MaxConnections: options.MaxSignalConnections,
MaxUnauthenticatedConnections: options.MaxUnauthenticatedSignalConnections,
AfterFunc: options.AfterFunc,
@@ -189,7 +187,7 @@ func New(options Options) (*Server, error) {
// newRoomStore builds the room store createScreenerServer would otherwise take
// from Options.RoomStore. An empty RoomDatabasePath is the memory-only mode.
func newRoomStore(options Options, now func() int64) (*room.Store, error) {
func newRoomStore(options Options) (*room.Store, error) {
if options.RoomStore != nil {
return options.RoomStore, nil
}
@@ -204,11 +202,9 @@ func newRoomStore(options Options, now func() int64) (*room.Store, error) {
database = opened
}
return room.New(room.Options{
LeaseMs: options.Config.RoomLeaseMs,
MaxRooms: room.Capacity,
MaxViewersPerRoom: options.Config.MaxViewersPerRoom,
Database: database,
Now: now,
})
}
internal/server/app/app_test.go
+11 -42
View File
@@ -58,7 +58,6 @@ func testConfig(t *testing.T) config.Config {
PublicBaseURL: publicBaseURL,
AllowedOrigins: map[string]struct{}{allowedOrigin: {}},
SiteAccessPassword: testAccessPassword,
RoomLeaseMs: 86_400_000,
MaxViewersPerRoom: 8,
EndpointMediaCopyCapacity: 2,
NATPredictionEnabled: false,
@@ -66,7 +65,7 @@ func testConfig(t *testing.T) config.Config {
}
// newServer builds a server without listening. It mirrors the TS harness: the
// heartbeat and cleanup intervals are pushed out of the way and a fake room
// heartbeat interval is pushed out of the way and a fake room
// control is injected whenever a LiveKit fallback is configured.
func newServer(t *testing.T, options Options) *Server {
t.Helper()
@@ -76,9 +75,6 @@ func newServer(t *testing.T, options Options) *Server {
if options.HeartbeatIntervalMs == 0 {
options.HeartbeatIntervalMs = 60_000
}
if options.CleanupIntervalMs == 0 {
options.CleanupIntervalMs = 60_000
}
server, err := New(options)
if err != nil {
t.Fatalf("New: %v", err)
@@ -734,18 +730,12 @@ func TestRoomCreationIssuesAnIndependentFragmentOnlyViewerGrant(t *testing.T) {
if created.CodeEntryPolicy != "open" {
t.Fatalf("codeEntryPolicy = %q", created.CodeEntryPolicy)
}
if created.ExpiresAt == nil || *created.ExpiresAt == "" {
t.Fatal("expiresAt is missing")
}
if created.RoomLeaseSeconds != 86_400 {
t.Fatalf("roomLeaseSeconds = %d", created.RoomLeaseSeconds)
}
// The response carries no viewer grant expiry and no ICE configuration.
// The response carries only room authority and no ICE configuration.
var raw map[string]json.RawMessage
if err := json.Unmarshal([]byte(response.body), &raw); err != nil {
t.Fatalf("decode body: %v", err)
}
for _, absent := range []string{"viewerGrantExpiresAt", "iceConfig", "viewerGrant"} {
for _, absent := range []string{"iceConfig", "viewerGrant"} {
if _, present := raw[absent]; present {
t.Fatalf("response exposes %q", absent)
}
@@ -818,16 +808,10 @@ func TestRoomCreationAllocatesUniqueFourDigitCodesConcurrently(t *testing.T) {
if created[0].RoomID == created[1].RoomID {
t.Fatalf("both rooms took the code %q", created[0].RoomID)
}
for _, room := range created {
if room.ExpiresAt == nil {
t.Fatal("expiresAt is null")
}
}
}
func TestRoomCreationReusesAFreePreferredCodeAndNeverReplacesAnOccupiedRoom(t *testing.T) {
configuration := testConfig(t)
configuration.RoomLeaseMs = 90_000
server := start(t, Options{Config: configuration})
cookie := server.cookie()
@@ -839,9 +823,6 @@ func TestRoomCreationReusesAFreePreferredCodeAndNeverReplacesAnOccupiedRoom(t *t
if preferred.RoomID != "4321" {
t.Fatalf("preferred roomId = %q", preferred.RoomID)
}
if preferred.RoomLeaseSeconds != 90 {
t.Fatalf("roomLeaseSeconds = %d", preferred.RoomLeaseSeconds)
}
if fallback.RoomID == "4321" {
t.Fatal("the occupied preferred code was handed out twice")
}
@@ -933,7 +914,6 @@ func TestRoomMutationFailuresMapToTheHTTPTable(t *testing.T) {
{"replacement busy", replacementFailure, &room.Error{Code: room.CodeRoomBusy}, 503, "Room replacement unavailable", true},
{"replacement token", replacementFailure, &room.Error{Code: room.CodeInvalidToken}, 404, "Room not found", true},
{"replacement missing", replacementFailure, &room.Error{Code: room.CodeRoomNotFound}, 404, "Room not found", true},
{"replacement expired", replacementFailure, &room.Error{Code: room.CodeRoomExpired}, 404, "Room not found", true},
{"replacement full", replacementFailure, &room.Error{Code: room.CodeRoomFull}, 0, "", false},
{"replacement other", replacementFailure, otherError, 0, "", false},
{"access busy", accessFailure, &room.Error{Code: room.CodeRoomBusy}, 503, "Room access update unavailable", true},
@@ -965,10 +945,9 @@ func TestRoomMutationFailuresMapToTheHTTPTable(t *testing.T) {
}
}
func TestRoomAccessManagesADormantRoomWithoutStartingSharingOrRenewing(t *testing.T) {
func TestRoomAccessManagesADormantRoomWithoutStartingSharing(t *testing.T) {
var now atomic.Int64
configuration := testConfig(t)
configuration.RoomLeaseMs = 1_000
server := start(t, Options{Config: configuration, Now: now.Load})
cookie := server.cookie()
created := server.createRoom(roomRequest{cookie: cookie}).
@@ -1024,20 +1003,11 @@ func TestRoomAccessManagesADormantRoomWithoutStartingSharingOrRenewing(t *testin
t.Fatal("managing access started a sharing session")
}
now.Store(1_001)
expired, err := server.Store().ExpireRooms(1_001)
if err != nil {
t.Fatalf("ExpireRooms: %v", err)
}
found := false
for _, closed := range expired {
if closed.RoomID == created.RoomID {
found = true
}
}
if !found {
t.Fatalf("access management renewed the lease; expired = %v", expired)
}
now.Store(30 * 24 * 60 * 60 * 1_000)
server.updateRoomAccess(accessRequest{
roomID: created.RoomID, hostToken: created.HostToken, cookie: server.cookie(),
body: `{"action":"set-code-entry-policy","policy":"open"}`,
}).expectStatus(http.StatusOK)
}
func TestRoomAccessRequiresSameOriginSiteAccessAndTheExactHostToken(t *testing.T) {
@@ -1112,7 +1082,7 @@ func TestRoomCreationReturnsServiceUnavailableAtTheGlobalRoomBound(t *testing.T)
configuration := testConfig(t)
configuration.SiteAccessPassword = ""
store, err := room.New(room.Options{
LeaseMs: 86_400_000, MaxRooms: 1, MaxViewersPerRoom: 8,
MaxRooms: 1, MaxViewersPerRoom: 8,
})
if err != nil {
t.Fatalf("room.New: %v", err)
@@ -1173,7 +1143,6 @@ func TestRestoresStableRoomAuthorityAcrossAnApplicationRestart(t *testing.T) {
configuration := testConfig(t)
configuration.SiteAccessPassword = ""
configuration.RoomDatabasePath = databasePath
configuration.RoomLeaseMs = 1_000
first := start(t, Options{Config: configuration, Now: now.Load})
created := first.createRoom(roomRequest{codeEntryPolicy: "private"}).
@@ -1182,7 +1151,7 @@ func TestRestoresStableRoomAuthorityAcrossAnApplicationRestart(t *testing.T) {
t.Fatalf("close first: %v", err)
}
now.Store(200)
now.Store(30 * 24 * 60 * 60 * 1_000)
second := start(t, Options{Config: configuration, Now: now.Load})
if size := second.Store().Size(); size != 1 {
t.Fatalf("restored room count = %d, want 1", size)
internal/server/app/routes.go
+1 -5
View File
@@ -371,10 +371,6 @@ func (s *Server) createRoomResponse(created room.CreatedRoom) protocol.CreateRoo
HostToken: created.HostToken,
InviteURL: signal.InviteURL(s.config.PublicBaseURL, created.RoomID, created.ViewerGrant),
CodeEntryPolicy: created.CodeEntryPolicy,
ExpiresAt: created.ExpiresAt,
// The lease is configured in whole seconds; TS divided by 1000 into a
// JS number and the response schema declares an integer.
RoomLeaseSeconds: protocol.Int(s.config.RoomLeaseMs / 1_000),
}
}
@@ -422,7 +418,7 @@ func replacementFailure(err error) (int, string, bool) {
return http.StatusServiceUnavailable, "Room capacity reached", true
case room.CodeRoomAccessDenied, room.CodeRoomBusy:
return http.StatusServiceUnavailable, "Room replacement unavailable", true
case room.CodeInvalidToken, room.CodeRoomNotFound, room.CodeRoomExpired:
case room.CodeInvalidToken, room.CodeRoomNotFound:
return http.StatusNotFound, "Room not found", true
}
return 0, "", false
internal/server/config/config.go
+11 -18
View File
@@ -1,7 +1,7 @@
// Package config ports src/server/config.ts, src/server/local-config.ts and
// src/server/ice.ts: the Hosted environment contract, the Client's local
// composition, and the ICE configuration derived from either. It performs no
// I/O; the caller supplies the environment as a map.
// composition, and the ICE configuration derived from either. The caller
// supplies the environment as a map; loading opens no runtime resources.
package config
import (
@@ -33,7 +33,6 @@ const (
minSiteAccessPasswordBytes = 8
maxSiteAccessPasswordBytes = 128
defaultMaxViewersPerRoom = 8
defaultRoomLeaseSeconds = 86_400
defaultPort = 8787
maxPort = 65_535
)
@@ -62,7 +61,8 @@ var removedEnvironmentVariables = []struct{ name, reason string }{
{"PEER_ASSISTED_MEDIA", "peer-assisted media is always enabled"},
{"HOST_ADMISSION_PASSWORD", "use SITE_ACCESS_PASSWORD"},
{"MAX_PEER_RELAY_DOWNSTREAM_EDGES", "use ENDPOINT_MEDIA_COPY_CAPACITY"},
{"ROOM_TTL_SECONDS", "use ROOM_LEASE_SECONDS"},
{"ROOM_TTL_SECONDS", "rooms do not expire"},
{"ROOM_LEASE_SECONDS", "rooms do not expire"},
{"ACCESS_PASSWORD", "use SITE_ACCESS_PASSWORD"},
{"NODE_ENV", "use SCREENER_ENV"},
}
@@ -88,7 +88,6 @@ type Config struct {
AllowedOrigins map[string]struct{}
SiteAccessPassword string
RoomDatabasePath string
RoomLeaseMs int64
MaxViewersPerRoom int
EndpointMediaCopyCapacity int
SFU *SFUConfig
@@ -208,17 +207,10 @@ func Load(env map[string]string) (Config, error) {
return Config{}, errors.New("STUN is required in production")
}
// ALLOWED_ORIGINS and ROOM_LEASE_SECONDS are validated inside the returned
// object literal, i.e. after every check above; keep that order.
allowedOrigins, err := parseOrigins(env["ALLOWED_ORIGINS"], Origin(publicBaseURL))
if err != nil {
return Config{}, err
}
roomLeaseSeconds, err := parsePositiveInteger(env["ROOM_LEASE_SECONDS"],
defaultRoomLeaseSeconds, "ROOM_LEASE_SECONDS")
if err != nil {
return Config{}, err
}
return Config{
Env: environment,
@@ -228,7 +220,6 @@ func Load(env map[string]string) (Config, error) {
AllowedOrigins: allowedOrigins,
SiteAccessPassword: siteAccessPassword,
RoomDatabasePath: roomDatabasePath,
RoomLeaseMs: roomLeaseSeconds * 1_000,
MaxViewersPerRoom: int(maxViewersPerRoom),
EndpointMediaCopyCapacity: int(endpointMediaCopyCapacity),
SFU: sfu,
@@ -288,16 +279,18 @@ func parseBoundedInteger(value string, fallback int64, name string, minimum, max
return parsed, nil
}
// parseRoomDatabasePath ports parseRoomDatabasePath. DECISIONS D8: absoluteness
// is filepath.IsAbs, which differs from Node's path.isAbsolute on Windows,
// where "/var/lib/screener/rooms.sqlite" is relative to the current volume and
// is therefore rejected here. Hosted runs on Linux.
// Hosted defaults to durable room authority in its working directory. Explicit
// paths stay absolute so a deployment cannot relocate an existing database
// when its working directory changes. :memory: selects the in-memory RoomStore.
func parseRoomDatabasePath(value string) (string, error) {
path := strings.TrimFunc(value, protocol.IsJSWhitespace)
if path == "" {
return filepath.Abs("rooms.sqlite")
}
if path == ":memory:" {
return "", nil
}
if path == ":memory:" || strings.ContainsRune(path, 0) || !filepath.IsAbs(path) {
if strings.ContainsRune(path, 0) || !filepath.IsAbs(path) {
return "", errors.New("ROOM_DATABASE_PATH must be an absolute file path")
}
return path, nil
internal/server/config/config_test.go
+35 -32
View File
@@ -3,6 +3,7 @@ package config
import (
"fmt"
"maps"
"os"
"path/filepath"
"reflect"
"slices"
@@ -64,7 +65,11 @@ func TestLoadDevelopmentDefaults(t *testing.T) {
if !reflect.DeepEqual(config.AllowedOrigins, want) {
t.Errorf("AllowedOrigins = %v", config.AllowedOrigins)
}
if config.RoomDatabasePath != "" {
defaultDatabasePath, err := filepath.Abs("rooms.sqlite")
if err != nil {
t.Fatal(err)
}
if config.RoomDatabasePath != defaultDatabasePath {
t.Errorf("RoomDatabasePath = %q", config.RoomDatabasePath)
}
if len(config.STUNURLs) != 0 {
@@ -88,14 +93,35 @@ func TestLoadDevelopmentDefaults(t *testing.T) {
if config.SFU != nil {
t.Errorf("SFU = %+v", config.SFU)
}
if config.RoomLeaseMs != 86_400_000 {
t.Errorf("RoomLeaseMs = %d", config.RoomLeaseMs)
}
if config.SiteAccessPassword != "" {
t.Error("SiteAccessPassword is set")
}
}
func TestLoadRoomStorage(t *testing.T) {
directory := t.TempDir()
t.Chdir(directory)
for _, base := range []map[string]string{nil, productionBase} {
for _, override := range []map[string]string{nil,
{"ROOM_DATABASE_PATH": ""}, {"ROOM_DATABASE_PATH": " \t"},
{"ROOM_DATABASE_PATH": ":memory:"},
} {
configuration := mustLoad(t, env(base, override))
want := filepath.Join(directory, "rooms.sqlite")
if override["ROOM_DATABASE_PATH"] == ":memory:" {
want = ""
}
if configuration.RoomDatabasePath != want {
t.Errorf("Load(%v): RoomDatabasePath = %q, want %q", override, configuration.RoomDatabasePath, want)
}
}
}
entries, err := os.ReadDir(directory)
if err != nil || len(entries) != 0 {
t.Fatalf("configuration loading created runtime files: %v, %v", entries, err)
}
}
func TestLoadSFU(t *testing.T) {
config := mustLoad(t, map[string]string{
"LISTEN_HOST": "127.0.0.1",
@@ -231,17 +257,6 @@ func TestLoadAccepts(t *testing.T) {
t.Error("NATPredictionEnabled = false")
}
}},
{"room lease default", map[string]string{}, func(t *testing.T, c Config) {
if c.RoomLeaseMs != 86_400_000 {
t.Errorf("RoomLeaseMs = %d", c.RoomLeaseMs)
}
}},
{"room lease override", map[string]string{"ROOM_LEASE_SECONDS": "3600"},
func(t *testing.T, c Config) {
if c.RoomLeaseMs != 3_600_000 {
t.Errorf("RoomLeaseMs = %d", c.RoomLeaseMs)
}
}},
{"file-backed room authority in development",
map[string]string{"ROOM_DATABASE_PATH": " " + developmentPath + " "},
func(t *testing.T, c Config) {
@@ -249,12 +264,6 @@ func TestLoadAccepts(t *testing.T) {
t.Errorf("RoomDatabasePath = %q", c.RoomDatabasePath)
}
}},
{"blank room database path", map[string]string{"ROOM_DATABASE_PATH": ""},
func(t *testing.T, c Config) {
if c.RoomDatabasePath != "" {
t.Errorf("RoomDatabasePath = %q", c.RoomDatabasePath)
}
}},
{"file-backed room authority in production",
env(productionBase, map[string]string{"ROOM_DATABASE_PATH": productionPath}),
func(t *testing.T, c Config) {
@@ -282,12 +291,6 @@ func TestLoadAccepts(t *testing.T) {
t.Errorf("Port = %d", c.Port)
}
}},
{"exponent room lease", map[string]string{"ROOM_LEASE_SECONDS": "1e3"},
func(t *testing.T, c Config) {
if c.RoomLeaseMs != 1_000_000 {
t.Errorf("RoomLeaseMs = %d", c.RoomLeaseMs)
}
}},
}
for _, testCase := range cases {
@@ -340,8 +343,6 @@ func TestLoadRejects(t *testing.T) {
"ENDPOINT_MEDIA_COPY_CAPACITY must be between 1 and 3"},
{"fractional copy capacity", map[string]string{"ENDPOINT_MEDIA_COPY_CAPACITY": "1.5"},
"ENDPOINT_MEDIA_COPY_CAPACITY must be a positive integer"},
{"zero room lease", map[string]string{"ROOM_LEASE_SECONDS": "0"},
"ROOM_LEASE_SECONDS must be a positive integer"},
{"port zero", map[string]string{"PORT": "0"}, "PORT must be a positive integer"},
{"port not a number", map[string]string{"PORT": "eight"}, "PORT must be a positive integer"},
{"port above the TCP ceiling", map[string]string{"PORT": "65536"}, "PORT must be at most 65535"},
@@ -434,8 +435,6 @@ func TestLoadRejects(t *testing.T) {
"NAT_PREDICTION_ENABLED must be true or false"},
// Room database path.
{"memory room database path", map[string]string{"ROOM_DATABASE_PATH": ":memory:"},
"ROOM_DATABASE_PATH must be an absolute file path"},
{"room database path with a NUL", map[string]string{"ROOM_DATABASE_PATH": "rooms\x00.sqlite"},
"ROOM_DATABASE_PATH must be an absolute file path"},
{"bare room database file name", map[string]string{"ROOM_DATABASE_PATH": "rooms.sqlite"},
@@ -461,7 +460,11 @@ func TestLoadRejects(t *testing.T) {
{"removed peer-assisted toggle", map[string]string{"PEER_ASSISTED_MEDIA": "true"},
"PEER_ASSISTED_MEDIA is no longer supported; peer-assisted media is always enabled"},
{"removed room TTL", map[string]string{"ROOM_TTL_SECONDS": ""},
"ROOM_TTL_SECONDS is no longer supported; use ROOM_LEASE_SECONDS"},
"ROOM_TTL_SECONDS is no longer supported; rooms do not expire"},
{"removed room lease", map[string]string{"ROOM_LEASE_SECONDS": "86400"},
"ROOM_LEASE_SECONDS is no longer supported; rooms do not expire"},
{"blank removed room lease", map[string]string{"ROOM_LEASE_SECONDS": ""},
"ROOM_LEASE_SECONDS is no longer supported; rooms do not expire"},
{"removed access password", map[string]string{"ACCESS_PASSWORD": "legacy-password"},
"ACCESS_PASSWORD is no longer supported; use SITE_ACCESS_PASSWORD"},
{"blank access password", map[string]string{"ACCESS_PASSWORD": ""},
internal/server/config/local.go
-4
View File
@@ -12,9 +12,6 @@ import (
"github.com/TNTcraftHIM/Screener/internal/server/protocol"
)
// Ported from src/server/local-config.ts.
const localRoomLeaseMs = 24 * 60 * 60 * 1_000
// localPasswordPattern is VISIBLE_ASCII_PATTERN from local-config.ts. Because
// it admits one byte per character the TypeScript's extra Buffer.byteLength
// bounds are already implied by {8,128}.
@@ -109,7 +106,6 @@ func Local(options LocalOptions) (Config, error) {
PublicBaseURL: publicBaseURL,
AllowedOrigins: allowedOrigins,
SiteAccessPassword: siteAccessPassword,
RoomLeaseMs: localRoomLeaseMs,
MaxViewersPerRoom: protocol.MaxViewersPerRoomLimit,
EndpointMediaCopyCapacity: protocol.DefaultEndpointMediaCopyCapacity,
STUNURLs: stunURLs,
internal/server/config/local_test.go
-3
View File
@@ -40,9 +40,6 @@ func TestLocalMapsClientInputs(t *testing.T) {
if config.SiteAccessPassword != "local-access-password" {
t.Errorf("SiteAccessPassword = %q", config.SiteAccessPassword)
}
if config.RoomLeaseMs != 86_400_000 {
t.Errorf("RoomLeaseMs = %d", config.RoomLeaseMs)
}
if config.MaxViewersPerRoom != 20 {
t.Errorf("MaxViewersPerRoom = %d", config.MaxViewersPerRoom)
}
internal/server/protocol/constants.go
+1 -1
View File
@@ -13,7 +13,7 @@ const (
MaxViewersPerRoomLimit = 20
MaxParticipantsPerRoomLimit = MaxViewersPerRoomLimit + 1
MaxSignalBytes = 64 * 1024
SignalingProtocol = "screener-v22"
SignalingProtocol = "screener-v23"
RoomCodeLength = 4
MaxMediaRouteRevision = MaxSafeInteger
MaxIceServerURLs = 8
internal/server/protocol/contract_test.go
+4 -8
View File
@@ -75,9 +75,8 @@ const (
func authenticatedHost() []member {
return []member{
{"type", `"authenticated"`},
{"protocol", `"screener-v22"`},
{"protocol", `"screener-v23"`},
{"peerId", `"host_12345678"`},
{"roomExpiresAt", `"2026-08-18T18:00:00.000Z"`},
{"maxViewers", `8`},
{"endpointMediaCopyCapacity", `2`},
{"hostOnline", `true`},
@@ -164,10 +163,7 @@ func TestAuthenticatedPasswordStateIsHostScoped(t *testing.T) {
object(with(viewer, member{"hostPaused", `true`})), true)
}
// "represents persistent rooms without a room expiry" and the revision bound.
func TestAuthenticatedRoomLifetimeAndRevision(t *testing.T) {
assertServerMessage(t, "no expiry",
object(with(authenticatedHost(), member{"roomExpiresAt", `null`})), true)
func TestAuthenticatedRoomRevision(t *testing.T) {
assertServerMessage(t, "revision at the ceiling",
object(with(authenticatedHost(), member{"routeRevision", `9007199254740991`})), true)
assertServerMessage(t, "revision past the ceiling",
@@ -584,7 +580,7 @@ func TestAuthenticatedKeyOrder(t *testing.T) {
}
}
want := []string{
"type", "protocol", "peerId", "roomExpiresAt", "maxViewers",
"type", "protocol", "peerId", "maxViewers",
"endpointMediaCopyCapacity", "hostOnline", "hostPaused", "connectionId",
"iceConfig", "routePolicy", "codeEntryPolicy", "viewerAuthorizationGeneration",
"mediaMode", "shareGeneration", "routeRevision", "routeAssignment",
@@ -622,7 +618,7 @@ func TestOptionalAndNullableAreExact(t *testing.T) {
// A host authenticate without routePolicy takes DEFAULT_ROUTE_POLICY.
func TestAuthenticateAppliesTheRoutePolicyDefault(t *testing.T) {
message, err := DecodeClientMessage([]byte(
`{"type":"authenticate","protocol":"screener-v22","roomId":"1234","role":"host",` +
`{"type":"authenticate","protocol":"screener-v23","roomId":"1234","role":"host",` +
`"token":"` + repeat("a", 43) + `","clientId":"client_12345678"}`))
if err != nil {
t.Fatal(err)
internal/server/protocol/http.go
+4 -6
View File
@@ -158,12 +158,10 @@ func DecodeRoomAccessUpdateRequest(data []byte) (RoomAccessUpdateRequest, error)
// CreateRoomResponse mirrors createRoomResponseSchema.
type CreateRoomResponse struct {
RoomID string `json:"roomId"`
HostToken string `json:"hostToken"`
InviteURL string `json:"inviteUrl"`
CodeEntryPolicy CodeEntryPolicy `json:"codeEntryPolicy"`
ExpiresAt *string `json:"expiresAt"`
RoomLeaseSeconds Int `json:"roomLeaseSeconds"`
RoomID string `json:"roomId"`
HostToken string `json:"hostToken"`
InviteURL string `json:"inviteUrl"`
CodeEntryPolicy CodeEntryPolicy `json:"codeEntryPolicy"`
}
// RoomAccessUpdateResponse is one member of roomAccessUpdateResponseSchema.
internal/server/protocol/scalars.go
-10
View File
@@ -77,16 +77,6 @@ func ValidViewerPassword(value string) bool {
viewerPasswordPattern.MatchString(value)
}
// isoDateTimePattern mirrors z.string().datetime(): an RFC 3339 UTC instant
// with a "Z" designator and no offset.
var isoDateTimePattern = regexp.MustCompile(
`^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d+)?Z$`)
// ValidISODateTime reports whether value is a zod `.datetime()` string.
func ValidISODateTime(value string) bool {
return isoDateTimePattern.MatchString(value)
}
// forbiddenDisplayNameCharacters is FORBIDDEN_DISPLAY_NAME_CHARACTERS from
// src/shared/protocol.ts.
var forbiddenDisplayNameCharacters = regexp.MustCompile(
internal/server/protocol/server.go
+5 -10
View File
@@ -19,7 +19,6 @@ type AuthenticatedHostMessage struct {
Type string `json:"type"`
Protocol string `json:"protocol"`
PeerID string `json:"peerId"`
RoomExpiresAt *string `json:"roomExpiresAt"`
MaxViewers Int `json:"maxViewers"`
EndpointMediaCopyCapacity Int `json:"endpointMediaCopyCapacity"`
HostOnline bool `json:"hostOnline"`
@@ -44,7 +43,6 @@ type AuthenticatedViewerMessage struct {
Type string `json:"type"`
Protocol string `json:"protocol"`
PeerID string `json:"peerId"`
RoomExpiresAt *string `json:"roomExpiresAt"`
MaxViewers Int `json:"maxViewers"`
EndpointMediaCopyCapacity Int `json:"endpointMediaCopyCapacity"`
HostOnline bool `json:"hostOnline"`
@@ -294,7 +292,7 @@ func decodeEmptyServerMessage(data []byte, message ServerMessage) (ServerMessage
func validateAuthenticatedBase(
present fields,
protocol string, peerID string, roomExpiresAt *string,
protocol string, peerID string,
maxViewers, endpointCapacity Int, connectionID *string,
codeEntryPolicy, viewerAuthorizationGeneration, mediaMode string,
shareGeneration *string, routeRevision Int,
@@ -306,7 +304,7 @@ func validateAuthenticatedBase(
return err
}
if err := present.requireNullable(
"roomExpiresAt", "connectionId", "shareGeneration"); err != nil {
"connectionId", "shareGeneration"); err != nil {
return err
}
if err := present.optional("hostPaused", "routePolicy"); err != nil {
@@ -318,9 +316,6 @@ func validateAuthenticatedBase(
if !ValidOpaqueID(peerID) {
return errors.New("peerId is not an opaque id")
}
if roomExpiresAt != nil && !ValidISODateTime(*roomExpiresAt) {
return errors.New("roomExpiresAt is not an ISO instant")
}
if !inRangeInt(maxViewers, 1, MaxViewersPerRoomLimit) {
return errors.New("maxViewers is out of range")
}
@@ -364,7 +359,7 @@ func decodeAuthenticated(data []byte) (ServerMessage, error) {
return nil, err
}
if err := validateAuthenticatedBase(present, message.Protocol, message.PeerID,
message.RoomExpiresAt, message.MaxViewers, message.EndpointMediaCopyCapacity,
message.MaxViewers, message.EndpointMediaCopyCapacity,
message.ConnectionID, message.CodeEntryPolicy,
message.ViewerAuthorizationGeneration, message.MediaMode,
message.ShareGeneration, message.RouteRevision); err != nil {
@@ -381,7 +376,7 @@ func decodeAuthenticated(data []byte) (ServerMessage, error) {
return nil, err
}
if err := validateAuthenticatedBase(present, message.Protocol, message.PeerID,
message.RoomExpiresAt, message.MaxViewers, message.EndpointMediaCopyCapacity,
message.MaxViewers, message.EndpointMediaCopyCapacity,
message.ConnectionID, message.CodeEntryPolicy,
message.ViewerAuthorizationGeneration, message.MediaMode,
message.ShareGeneration, message.RouteRevision); err != nil {
@@ -653,7 +648,7 @@ func decodeRoomClosed(data []byte) (ServerMessage, error) {
if err := present.require("type", "reason"); err != nil {
return nil, err
}
if !enumOf(message.Reason, "host-ended", "expired") {
if message.Reason != "host-ended" {
return nil, fmt.Errorf("unknown room-closed reason %q", message.Reason)
}
return message, nil
internal/server/protocol/types.go
+1 -1
View File
@@ -45,7 +45,7 @@ var (
routeDiagnosticStages = []string{"admission", "first-frame", "quality-proof"}
senderQualityStates = []string{"unknown", "healthy", "degraded"}
senderQualityDegradedReasons = []string{"none", "bandwidth", "cpu"}
errorCodes = []string{"AUTH_REQUIRED", "INVALID_MESSAGE", "INVALID_TOKEN", "ROOM_ACCESS_DENIED", "ROOM_NOT_FOUND", "ROOM_EXPIRED", "ROOM_FULL", "HOST_ALREADY_CONNECTED", "PEER_NOT_FOUND", "FORBIDDEN", "SERVER_ERROR"}
errorCodes = []string{"AUTH_REQUIRED", "INVALID_MESSAGE", "INVALID_TOKEN", "ROOM_ACCESS_DENIED", "ROOM_NOT_FOUND", "ROOM_FULL", "HOST_ALREADY_CONNECTED", "PEER_NOT_FOUND", "FORBIDDEN", "SERVER_ERROR"}
)
func validRevision(revision Int) bool {
internal/server/room/compat_test.go
-155
View File
@@ -1,155 +0,0 @@
package room
import (
"bytes"
"os"
"path/filepath"
"testing"
"github.com/TNTcraftHIM/Screener/internal/server/protocol"
)
// testdata/node-rooms.db was written once by node:sqlite (DatabaseSync) with
// the schema of src/server/room-database.ts and two synthetic rooms: 4321 is
// active (NULL lease) with a 48-byte password material, 5678 is dormant with a
// lease of 1000 ms. No real token, grant or password is involved. The extension
// is .db because .gitignore excludes *.sqlite repository-wide.
func copyNodeFixture(t *testing.T) string {
t.Helper()
fixture, err := os.ReadFile(filepath.Join("testdata", "node-rooms.db"))
if err != nil {
t.Fatalf("read fixture: %v", err)
}
path := filepath.Join(t.TempDir(), "rooms.sqlite")
if err := os.WriteFile(path, fixture, 0o600); err != nil {
t.Fatalf("write fixture copy: %v", err)
}
return path
}
func openFixture(t *testing.T, startupNowMs, activeLeaseExpiresAtMs int64) []StoredRoomAuthority {
t.Helper()
database, err := NewDatabase(copyNodeFixture(t))
if err != nil {
t.Fatalf("NewDatabase: %v", err)
}
t.Cleanup(func() { _ = database.Close() })
restored, err := database.Initialize(startupNowMs, activeLeaseExpiresAtMs)
if err != nil {
t.Fatalf("Initialize: %v", err)
}
return restored
}
func TestRecoversANodeWrittenDatabase(t *testing.T) {
restored := openFixture(t, 5_000, 6_000)
if len(restored) != 1 {
t.Fatalf("restored %d rooms, want 1 (the dormant room was due at 1000)", len(restored))
}
room := restored[0]
if room.RoomID != "4321" || room.CodeEntryPolicy != protocol.CodeEntryPrivate {
t.Fatalf("restored %+v", room)
}
if room.LeaseExpiresAtMs == nil || *room.LeaseExpiresAtMs != 6_000 {
t.Fatalf("active room lease = %v, want the startup lease 6000", room.LeaseExpiresAtMs)
}
if !bytes.Equal(room.HostTokenDigest, bytes.Repeat([]byte{0x11}, 32)) ||
!bytes.Equal(room.ViewerGrantDigest, bytes.Repeat([]byte{0x22}, 32)) {
t.Fatal("digests did not survive the Node round trip")
}
if room.ViewerAuthorizationGeneration != "MzMzMzMzMzMzMzMzMzMzMw" {
t.Fatalf("generation = %q", room.ViewerAuthorizationGeneration)
}
if !bytes.Equal(room.ViewerPasswordMaterial, bytes.Repeat([]byte{0x44}, 48)) {
t.Fatalf("password material = %d bytes", len(room.ViewerPasswordMaterial))
}
}
func TestKeepsADormantNodeWrittenDeadline(t *testing.T) {
restored := openFixture(t, 500, 1_500)
if len(restored) != 2 {
t.Fatalf("restored %d rooms, want 2", len(restored))
}
// readStoredRooms orders by room_id.
if restored[0].RoomID != "4321" || restored[0].LeaseExpiresAtMs == nil ||
*restored[0].LeaseExpiresAtMs != 1_500 {
t.Fatalf("active room = %+v", restored[0])
}
if restored[1].RoomID != "5678" || restored[1].LeaseExpiresAtMs == nil ||
*restored[1].LeaseExpiresAtMs != 1_000 {
t.Fatalf("dormant room = %+v", restored[1])
}
if restored[1].ViewerPasswordMaterial != nil {
t.Fatalf("dormant room material = %v, want nil", restored[1].ViewerPasswordMaterial)
}
}
func TestStoreRestoresANodeWrittenDatabase(t *testing.T) {
testClock := &clock{nowMs: 5_000}
store := stableStore(t, copyNodeFixture(t), testClock, Options{LeaseMs: 1_000})
if store.Size() != 1 {
t.Fatalf("size = %d, want 1", store.Size())
}
// The recovered room holds the startup lease: 5000 + 1000.
testClock.nowMs = 6_001
expired, err := store.ExpireRooms(testClock.nowMs)
if err != nil {
t.Fatalf("ExpireRooms: %v", err)
}
if len(expired) != 1 || expired[0].RoomID != "4321" {
t.Fatalf("expired = %+v", expired)
}
}
// TestSchemaTextMatchesTheNodeFixture proves the CREATE TABLE text this package
// executes is byte for byte the one src/server/room-database.ts executed.
func TestSchemaTextMatchesTheNodeFixture(t *testing.T) {
path := databasePath(t)
database, err := NewDatabase(path)
if err != nil {
t.Fatalf("NewDatabase: %v", err)
}
if _, err := database.Initialize(0, 1_000); err != nil {
t.Fatalf("Initialize: %v", err)
}
if err := database.Close(); err != nil {
t.Fatalf("Close: %v", err)
}
created := readSchemaText(t, path)
fixture := readSchemaText(t, copyNodeFixture(t))
if created.sql != fixture.sql {
t.Fatalf("schema text differs\n go: %q\n node: %q", created.sql, fixture.sql)
}
if created.applicationID != fixture.applicationID ||
created.applicationID != roomDatabaseApplicationID {
t.Fatalf("application_id go=%d node=%d", created.applicationID, fixture.applicationID)
}
if created.userVersion != fixture.userVersion ||
created.userVersion != roomDatabaseSchemaVersion {
t.Fatalf("user_version go=%d node=%d", created.userVersion, fixture.userVersion)
}
}
type schemaText struct {
sql string
applicationID int64
userVersion int64
}
func readSchemaText(t *testing.T, path string) schemaText {
t.Helper()
raw := rawDatabase(t, path)
var text schemaText
if err := raw.QueryRow(
"SELECT sql FROM sqlite_schema WHERE name = 'rooms'").Scan(&text.sql); err != nil {
t.Fatalf("read schema: %v", err)
}
if err := raw.QueryRow("PRAGMA application_id").Scan(&text.applicationID); err != nil {
t.Fatalf("read application_id: %v", err)
}
if err := raw.QueryRow("PRAGMA user_version").Scan(&text.userVersion); err != nil {
t.Fatalf("read user_version: %v", err)
}
return text
}
internal/server/room/database.go
+23 -170
View File
@@ -15,18 +15,14 @@ import (
"github.com/TNTcraftHIM/Screener/internal/server/protocol"
)
// roomDatabaseApplicationID and roomDatabaseSchemaVersion are the
// APPLICATION_ID / SCHEMA_VERSION of src/server/room-database.ts, checked on
// open so another SQLite file cannot be adopted as the room authority. The ID
// is ASCII "SCRN"; roomSchema below repeats it in decimal because that literal
// has to match the TypeScript statement text byte for byte.
// Database identity and the single current schema are checked before adopting
// room authority. The application ID is ASCII "SCRN".
const (
roomDatabaseApplicationID = 0x5343524e
roomDatabaseSchemaVersion = 1
roomDatabaseSchemaVersion = 2
)
// roomSchema is the exact text src/server/room-database.ts executes, so a file
// created here and one created by the TypeScript hold the same sqlite_schema.
// roomSchema stores durable authority only, never presence or media state.
const roomSchema = `
CREATE TABLE rooms (
room_id TEXT PRIMARY KEY NOT NULL
@@ -46,15 +42,10 @@ const roomSchema = `
CHECK (
viewer_password_material IS NULL OR
(typeof(viewer_password_material) = 'blob' AND length(viewer_password_material) = 48)
),
lease_expires_at_ms INTEGER NULL
CHECK (
lease_expires_at_ms IS NULL OR
(lease_expires_at_ms > 0 AND lease_expires_at_ms <= 9007199254740991)
)
) STRICT;
PRAGMA application_id = 1396920910;
PRAGMA user_version = 1;
PRAGMA user_version = 2;
`
const selectStoredRooms = `SELECT
@@ -63,8 +54,7 @@ const selectStoredRooms = `SELECT
viewer_grant_digest,
viewer_authorization_generation,
code_entry_policy,
viewer_password_material,
lease_expires_at_ms
viewer_password_material
FROM rooms
ORDER BY room_id`
@@ -74,14 +64,12 @@ const insertStoredRoom = `INSERT INTO rooms (
viewer_grant_digest,
viewer_authorization_generation,
code_entry_policy,
viewer_password_material,
lease_expires_at_ms
) VALUES (?, ?, ?, ?, ?, ?, ?)`
viewer_password_material
) VALUES (?, ?, ?, ?, ?, ?)`
const deleteRoomStatement = "DELETE FROM rooms WHERE room_id = ? AND host_token_digest = ?"
// StoredRoomAuthority is StoredRoomAuthority. A nil ViewerPasswordMaterial and
// a nil LeaseExpiresAtMs are the TypeScript nulls.
// StoredRoomAuthority excludes transient sessions and presence.
type StoredRoomAuthority struct {
RoomID string
HostTokenDigest []byte
@@ -89,13 +77,6 @@ type StoredRoomAuthority struct {
ViewerAuthorizationGeneration string
CodeEntryPolicy protocol.CodeEntryPolicy
ViewerPasswordMaterial []byte
LeaseExpiresAtMs *int64
}
// RoomIdentity is the {roomId, hostTokenDigest} pair deleteRooms takes.
type RoomIdentity struct {
RoomID string
HostTokenDigest []byte
}
// Database is RoomDatabase: the SQLite stable authority. It holds one
@@ -119,25 +100,11 @@ func NewDatabase(path string) (*Database, error) {
}
// Initialize opens the file, takes the exclusive lock, creates or verifies the
// schema and recovers the stored rooms: rooms whose lease has passed are
// deleted and rooms that were active at shutdown take activeLeaseExpiresAtMs.
func (d *Database) Initialize(
startupNowMs, activeLeaseExpiresAtMs int64,
) (restored []StoredRoomAuthority, err error) {
// schema and restores all stored room authority.
func (d *Database) Initialize() (restored []StoredRoomAuthority, err error) {
if d.conn != nil {
return nil, errors.New("Room database is already initialized")
}
if err := assertTimestamp(startupNowMs, "Room database startup time", true); err != nil {
return nil, err
}
if err := assertTimestamp(
activeLeaseExpiresAtMs, "Room database active-room lease deadline", false); err != nil {
return nil, err
}
if activeLeaseExpiresAtMs <= startupNowMs {
return nil, errors.New(
"Room database active-room lease deadline must be in the future")
}
// `new DatabaseSync(this.path, { timeout: 0 })`: a lock conflict must fail
// immediately instead of waiting.
@@ -174,7 +141,7 @@ func (d *Database) Initialize(
if err = d.begin("BEGIN EXCLUSIVE"); err != nil {
return nil, err
}
restored, err = d.recover(startupNowMs, activeLeaseExpiresAtMs)
restored, err = d.recover()
if err != nil {
d.rollbackIfNeeded()
return nil, err
@@ -193,63 +160,14 @@ func uriPath(path string) string {
return strings.NewReplacer("%", "%25", "#", "%23", "?", "%3F").Replace(filepath.ToSlash(path))
}
func (d *Database) recover(
startupNowMs, activeLeaseExpiresAtMs int64,
) ([]StoredRoomAuthority, error) {
func (d *Database) recover() ([]StoredRoomAuthority, error) {
if err := d.ensureExactSchema(); err != nil {
return nil, err
}
if err := d.assertDatabaseIntegrity(); err != nil {
return nil, err
}
stored, err := d.readStoredRooms()
if err != nil {
return nil, err
}
expiredCount, activeCount := 0, 0
for _, room := range stored {
switch {
case room.LeaseExpiresAtMs == nil:
activeCount++
case *room.LeaseExpiresAtMs <= startupNowMs:
expiredCount++
}
}
deleted, err := d.exec(
`DELETE FROM rooms
WHERE lease_expires_at_ms IS NOT NULL
AND lease_expires_at_ms <= ?`, startupNowMs)
if err != nil {
return nil, err
}
if deleted != int64(expiredCount) {
return nil, errors.New("Room database expiry cleanup changed an unexpected row count")
}
converted, err := d.exec(
`UPDATE rooms
SET lease_expires_at_ms = ?
WHERE lease_expires_at_ms IS NULL`, activeLeaseExpiresAtMs)
if err != nil {
return nil, err
}
if converted != int64(activeCount) {
return nil, errors.New("Room database active-room recovery changed an unexpected row count")
}
restored := make([]StoredRoomAuthority, 0, len(stored))
for _, room := range stored {
if room.LeaseExpiresAtMs != nil && *room.LeaseExpiresAtMs <= startupNowMs {
continue
}
if room.LeaseExpiresAtMs == nil {
lease := activeLeaseExpiresAtMs
room.LeaseExpiresAtMs = &lease
}
restored = append(restored, room)
}
return restored, nil
return d.readStoredRooms()
}
// InsertRoom is insertRoom.
@@ -264,8 +182,7 @@ func (d *Database) InsertRoom(room StoredRoomAuthority) error {
room.ViewerGrantDigest,
room.ViewerAuthorizationGeneration,
room.CodeEntryPolicy,
room.ViewerPasswordMaterial,
room.LeaseExpiresAtMs)
room.ViewerPasswordMaterial)
if err != nil {
return err
}
@@ -273,31 +190,6 @@ func (d *Database) InsertRoom(room StoredRoomAuthority) error {
})
}
// SetLeaseDeadline is setLeaseDeadline; a nil deadline marks the room active.
func (d *Database) SetLeaseDeadline(
roomID string, hostTokenDigest []byte, leaseExpiresAtMs *int64,
) error {
if err := assertRoomIdentity(roomID, hostTokenDigest); err != nil {
return err
}
if leaseExpiresAtMs != nil {
if err := assertTimestamp(*leaseExpiresAtMs, "Room lease deadline", false); err != nil {
return err
}
}
return d.transaction(func() error {
changes, err := d.exec(
`UPDATE rooms
SET lease_expires_at_ms = ?
WHERE room_id = ? AND host_token_digest = ?`,
leaseExpiresAtMs, roomID, hostTokenDigest)
if err != nil {
return err
}
return assertSingleChange(changes, "lease update")
})
}
// SetViewerPassword is setViewerPassword; nil material clears the password.
func (d *Database) SetViewerPassword(
roomID string, hostTokenDigest, viewerPasswordMaterial []byte,
@@ -374,31 +266,15 @@ func (d *Database) SetViewerGrant(
// DeleteRoom is deleteRoom.
func (d *Database) DeleteRoom(roomID string, hostTokenDigest []byte) error {
return d.DeleteRooms([]RoomIdentity{{RoomID: roomID, HostTokenDigest: hostTokenDigest}})
}
// DeleteRooms is deleteRooms: one transaction, and any room whose authority no
// longer matches rolls the whole batch back.
func (d *Database) DeleteRooms(rooms []RoomIdentity) error {
for _, room := range rooms {
if err := assertRoomIdentity(room.RoomID, room.HostTokenDigest); err != nil {
return err
}
}
if len(rooms) == 0 {
return nil
if err := assertRoomIdentity(roomID, hostTokenDigest); err != nil {
return err
}
return d.transaction(func() error {
for _, room := range rooms {
changes, err := d.exec(deleteRoomStatement, room.RoomID, room.HostTokenDigest)
if err != nil {
return err
}
if err := assertSingleChange(changes, "delete"); err != nil {
return err
}
changes, err := d.exec(deleteRoomStatement, roomID, hostTokenDigest)
if err != nil {
return err
}
return nil
return assertSingleChange(changes, "delete")
})
}
@@ -423,8 +299,7 @@ func (d *Database) ReplaceRoom(
replacement.ViewerGrantDigest,
replacement.ViewerAuthorizationGeneration,
replacement.CodeEntryPolicy,
replacement.ViewerPasswordMaterial,
replacement.LeaseExpiresAtMs)
replacement.ViewerPasswordMaterial)
if err != nil {
return err
}
@@ -548,7 +423,6 @@ func (d *Database) assertExactColumns() error {
{"viewer_authorization_generation", "TEXT", 1, 0},
{"code_entry_policy", "TEXT", 1, 0},
{"viewer_password_material", "BLOB", 0, 0},
{"lease_expires_at_ms", "INTEGER", 0, 0},
}
rows, err := d.conn.QueryContext(context.Background(), "PRAGMA table_info(rooms)")
if err != nil {
@@ -597,7 +471,6 @@ func (d *Database) readStoredRooms() ([]StoredRoomAuthority, error) {
var (
room StoredRoomAuthority
policy string
lease sql.NullInt64
material []byte
)
if err := rows.Scan(
@@ -606,8 +479,7 @@ func (d *Database) readStoredRooms() ([]StoredRoomAuthority, error) {
&room.ViewerGrantDigest,
&room.ViewerAuthorizationGeneration,
&policy,
&material,
&lease); err != nil {
&material); err != nil {
return nil, err
}
if err := readDigest(room.HostTokenDigest, "Host token"); err != nil {
@@ -626,12 +498,6 @@ func (d *Database) readStoredRooms() ([]StoredRoomAuthority, error) {
}
room.ViewerPasswordMaterial = material
}
if lease.Valid {
if lease.Int64 <= 0 || lease.Int64 > protocol.MaxSafeInteger {
return nil, errors.New("Room database contains an invalid lease deadline")
}
room.LeaseExpiresAtMs = &lease.Int64
}
if err := assertStoredRoom(room); err != nil {
return nil, err
}
@@ -721,9 +587,6 @@ func assertStoredRoom(room StoredRoomAuthority) error {
if err := assertPasswordMaterial(room.ViewerPasswordMaterial); err != nil {
return err
}
if room.LeaseExpiresAtMs != nil {
return assertTimestamp(*room.LeaseExpiresAtMs, "Room database lease deadline", false)
}
return nil
}
@@ -764,16 +627,6 @@ func assertAuthorizationGeneration(value string) error {
return nil
}
// assertTimestamp is the TypeScript Number.isSafeInteger guard: SQLite STRICT
// accepts an integral REAL and a numeric TEXT in an INTEGER column, so the
// range check has to stay on this side.
func assertTimestamp(value int64, name string, allowZero bool) error {
if value > protocol.MaxSafeInteger || (allowZero && value < 0) || (!allowZero && value <= 0) {
return fmt.Errorf("%s must be a safe positive integer", name)
}
return nil
}
func assertSingleChange(changes int64, operation string) error {
if changes != 1 {
return fmt.Errorf("Room database %s did not match current authority", operation)
internal/server/room/database_test.go
+81 -172
View File
@@ -6,6 +6,7 @@ import (
"database/sql"
"os"
"path/filepath"
"reflect"
"strings"
"testing"
@@ -21,21 +22,17 @@ func databasePath(t *testing.T) string {
return filepath.Join(t.TempDir(), "rooms.sqlite")
}
func stableStore(t *testing.T, path string, testClock *clock, options Options) *Store {
func stableStore(t *testing.T, path string, options Options) *Store {
t.Helper()
database, err := NewDatabase(path)
if err != nil {
t.Fatalf("NewDatabase: %v", err)
}
if options.LeaseMs == 0 {
options.LeaseMs = 1_000
}
if options.MaxRooms == 0 {
options.MaxRooms = 8
}
options.MaxViewersPerRoom = 8
options.Database = database
options.Now = func() int64 { return testClock.nowMs }
store, err := New(options)
if err != nil {
t.Fatalf("New: %v", err)
@@ -72,8 +69,7 @@ func dbViewerInput(roomID, sessionID, viewerGrant string) ConnectParticipantInpu
func TestDoesNotPersistARoomWhenPasswordDerivationIsBusy(t *testing.T) {
path := databasePath(t)
testClock := &clock{nowMs: 1_000}
first := stableStore(t, path, testClock, Options{MaxRooms: 1})
first := stableStore(t, path, Options{MaxRooms: 1})
withSaturatedGate(t, first, func() {
_, err := tryCreateRoom(first, protocol.CodeEntryPrivate, "room-password", "4321")
@@ -86,7 +82,7 @@ func TestDoesNotPersistARoomWhenPasswordDerivationIsBusy(t *testing.T) {
t.Fatalf("Close: %v", err)
}
second := stableStore(t, path, testClock, Options{MaxRooms: 1})
second := stableStore(t, path, Options{MaxRooms: 1})
if second.Size() != 0 {
t.Fatalf("restored size = %d, want 0", second.Size())
}
@@ -97,8 +93,7 @@ func TestDoesNotPersistARoomWhenPasswordDerivationIsBusy(t *testing.T) {
func TestRestoresTheExactAuthorityAggregateWithoutParticipants(t *testing.T) {
path := databasePath(t)
testClock := &clock{nowMs: 1_000}
first := stableStore(t, path, testClock, Options{})
first := stableStore(t, path, Options{})
room := createRoom(t, first, protocol.CodeEntryOpen, "", "4321")
if _, err := first.SetCodeEntryPolicy(
room.RoomID, protocol.CodeEntryPrivate, room.HostToken); err != nil {
@@ -124,8 +119,7 @@ func TestRestoresTheExactAuthorityAggregateWithoutParticipants(t *testing.T) {
t.Fatalf("Close: %v", err)
}
testClock.nowMs = 2_000
second := stableStore(t, path, testClock, Options{})
second := stableStore(t, path, Options{})
if second.Size() != 1 {
t.Fatalf("restored size = %d, want 1", second.Size())
}
@@ -139,9 +133,6 @@ func TestRestoresTheExactAuthorityAggregateWithoutParticipants(t *testing.T) {
expectCode(t, err, CodeInvalidToken)
granted := mustConnect(t, second, dbViewerInput(room.RoomID, "current-grant", rotated.ViewerGrant))
if granted.ExpiresAt == nil || *granted.ExpiresAt != "1970-01-01T00:00:03.000Z" {
t.Fatalf("expiresAt = %v", granted.ExpiresAt)
}
if granted.ViewerAuthorizationGeneration != rotated.ViewerAuthorizationGeneration {
t.Fatalf("generation = %q, want %q",
granted.ViewerAuthorizationGeneration, rotated.ViewerAuthorizationGeneration)
@@ -157,10 +148,7 @@ func TestRestoresTheExactAuthorityAggregateWithoutParticipants(t *testing.T) {
}
_, err = second.ConnectParticipant(hostInput(room.RoomID, "wrong-token"))
expectCode(t, err, CodeInvalidToken)
resumed := mustConnect(t, second, hostInput(room.RoomID, room.HostToken, "recovered-host"))
if resumed.ExpiresAt != nil {
t.Fatalf("resumed expiresAt = %v, want null", *resumed.ExpiresAt)
}
mustConnect(t, second, hostInput(room.RoomID, room.HostToken, "recovered-host"))
if err := second.Close(); err != nil {
t.Fatalf("Close: %v", err)
}
@@ -178,7 +166,7 @@ func TestRestoresTheExactAuthorityAggregateWithoutParticipants(t *testing.T) {
want := []string{
"room_id", "host_token_digest", "viewer_grant_digest",
"viewer_authorization_generation", "code_entry_policy",
"viewer_password_material", "lease_expires_at_ms",
"viewer_password_material",
}
if strings.Join(columns, ",") != strings.Join(want, ",") {
t.Fatalf("columns = %v", columns)
@@ -187,16 +175,15 @@ func TestRestoresTheExactAuthorityAggregateWithoutParticipants(t *testing.T) {
t.Fatal("no stored room")
}
var (
roomID string
hostDigest []byte
grantDigest []byte
generation string
policy string
material []byte
leaseExpiresAtMs sql.NullInt64
roomID string
hostDigest []byte
grantDigest []byte
generation string
policy string
material []byte
)
if err := rows.Scan(&roomID, &hostDigest, &grantDigest, &generation, &policy,
&material, &leaseExpiresAtMs); err != nil {
&material); err != nil {
t.Fatalf("Scan: %v", err)
}
hostTokenDigest := sha256.Sum256([]byte(room.HostToken))
@@ -219,58 +206,11 @@ func TestRestoresTheExactAuthorityAggregateWithoutParticipants(t *testing.T) {
}
}
func TestConvertsAnActiveMarkerOnceAndPreservesDormantDeadlines(t *testing.T) {
func TestPreservesDisconnectedAuthorityAcrossRepeatedRestarts(t *testing.T) {
path := databasePath(t)
testClock := &clock{nowMs: 100}
first := stableStore(t, path, testClock, Options{})
room := createRoom(t, first, protocol.CodeEntryOpen, "", "5678")
mustConnect(t, first, hostInput(room.RoomID, room.HostToken))
if err := first.Close(); err != nil {
t.Fatalf("Close: %v", err)
}
testClock.nowMs = 500
second := stableStore(t, path, testClock, Options{})
firstRestart := mustConnect(t, second, dbViewerInput(room.RoomID, "first-restart", room.ViewerGrant))
if firstRestart.ExpiresAt == nil || *firstRestart.ExpiresAt != "1970-01-01T00:00:01.500Z" {
t.Fatalf("expiresAt = %v", firstRestart.ExpiresAt)
}
if err := second.Close(); err != nil {
t.Fatalf("Close: %v", err)
}
testClock.nowMs = 700
third := stableStore(t, path, testClock, Options{})
secondRestart := mustConnect(t, third, dbViewerInput(room.RoomID, "second-restart", room.ViewerGrant))
if secondRestart.ExpiresAt == nil || *secondRestart.ExpiresAt != "1970-01-01T00:00:01.500Z" {
t.Fatalf("expiresAt = %v", secondRestart.ExpiresAt)
}
if err := third.Close(); err != nil {
t.Fatalf("Close: %v", err)
}
testClock.nowMs = 1_501
fourth := stableStore(t, path, testClock, Options{})
if fourth.Size() != 0 {
t.Fatalf("size = %d, want 0", fourth.Size())
}
replacement := createRoom(t, fourth, protocol.CodeEntryOpen, "", room.RoomID)
if replacement.RoomID != room.RoomID {
t.Fatalf("replacement room = %q", replacement.RoomID)
}
_, err := fourth.ConnectParticipant(hostInput(room.RoomID, room.HostToken))
expectCode(t, err, CodeInvalidToken)
_, err = fourth.ConnectParticipant(dbViewerInput(room.RoomID, "stale-grant", room.ViewerGrant))
expectCode(t, err, CodeInvalidToken)
}
func TestPreservesTheExactHostDisconnectDeadline(t *testing.T) {
path := databasePath(t)
testClock := &clock{nowMs: 100}
first := stableStore(t, path, testClock, Options{})
first := stableStore(t, path, Options{})
room := createRoom(t, first, protocol.CodeEntryOpen, "", "5890")
host := mustConnect(t, first, hostInput(room.RoomID, room.HostToken))
testClock.nowMs = 250
if _, err := first.DisconnectParticipant(room.RoomID, host.PeerID, "host-session"); err != nil {
t.Fatalf("DisconnectParticipant: %v", err)
}
@@ -278,18 +218,25 @@ func TestPreservesTheExactHostDisconnectDeadline(t *testing.T) {
t.Fatalf("Close: %v", err)
}
testClock.nowMs = 500
second := stableStore(t, path, testClock, Options{})
dormant := mustConnect(t, second, dbViewerInput(room.RoomID, "dormant-viewer", room.ViewerGrant))
if dormant.ExpiresAt == nil || *dormant.ExpiresAt != "1970-01-01T00:00:01.250Z" {
t.Fatalf("expiresAt = %v", dormant.ExpiresAt)
for range 3 {
restored := stableStore(t, path, Options{})
if restored.Size() != 1 {
t.Fatalf("size = %d, want 1", restored.Size())
}
viewer := mustConnect(t, restored, dbViewerInput(room.RoomID, "dormant-viewer", room.ViewerGrant))
if viewer.HostOnline {
t.Fatal("restored room reported an online host")
}
mustConnect(t, restored, hostInput(room.RoomID, room.HostToken))
if err := restored.Close(); err != nil {
t.Fatalf("Close: %v", err)
}
}
}
func TestPersistsExplicitRoomDeletionBeforeRecyclingItsCode(t *testing.T) {
path := databasePath(t)
testClock := &clock{nowMs: 10}
first := stableStore(t, path, testClock, Options{})
first := stableStore(t, path, Options{})
room := createRoom(t, first, protocol.CodeEntryOpen, "", "6789")
closed, err := first.AbandonRoom(room.RoomID)
if err != nil || closed == nil || closed.RoomID != room.RoomID {
@@ -299,7 +246,7 @@ func TestPersistsExplicitRoomDeletionBeforeRecyclingItsCode(t *testing.T) {
t.Fatalf("Close: %v", err)
}
second := stableStore(t, path, testClock, Options{})
second := stableStore(t, path, Options{})
if second.Size() != 0 {
t.Fatalf("size = %d, want 0", second.Size())
}
@@ -312,48 +259,9 @@ func TestPersistsExplicitRoomDeletionBeforeRecyclingItsCode(t *testing.T) {
}
}
func TestRollsBackABatchExpiryWhenAnyAuthorityIsStale(t *testing.T) {
path := databasePath(t)
testClock := &clock{nowMs: 10}
first := stableStore(t, path, testClock, Options{})
createRoom(t, first, protocol.CodeEntryOpen, "", "4321")
createRoom(t, first, protocol.CodeEntryOpen, "", "5678")
if err := first.Close(); err != nil {
t.Fatalf("Close: %v", err)
}
database, err := NewDatabase(path)
if err != nil {
t.Fatalf("NewDatabase: %v", err)
}
restored, err := database.Initialize(testClock.nowMs, 1_010)
if err != nil {
t.Fatalf("Initialize: %v", err)
}
if len(restored) != 2 {
t.Fatalf("restored %d rooms, want 2", len(restored))
}
err = database.DeleteRooms([]RoomIdentity{
{RoomID: restored[0].RoomID, HostTokenDigest: restored[0].HostTokenDigest},
{RoomID: restored[1].RoomID, HostTokenDigest: make([]byte, 32)},
})
if err == nil || err.Error() != "Room database delete did not match current authority" {
t.Fatalf("DeleteRooms: %v", err)
}
if err := database.Close(); err != nil {
t.Fatalf("Close: %v", err)
}
second := stableStore(t, path, testClock, Options{})
if second.Size() != 2 {
t.Fatalf("size = %d, want 2", second.Size())
}
}
func TestPersistsRoomReplacementAsOneAuthorityTransition(t *testing.T) {
path := databasePath(t)
testClock := &clock{nowMs: 10}
first := stableStore(t, path, testClock, Options{})
first := stableStore(t, path, Options{})
original := createRoom(t, first, protocol.CodeEntryOpen, "", "4321")
replacement, err := tryReplaceRoom(
first, original.RoomID, original.HostToken, protocol.CodeEntryPrivate, "new-password")
@@ -364,7 +272,7 @@ func TestPersistsRoomReplacementAsOneAuthorityTransition(t *testing.T) {
t.Fatalf("Close: %v", err)
}
second := stableStore(t, path, testClock, Options{})
second := stableStore(t, path, Options{})
if second.Size() != 1 {
t.Fatalf("size = %d, want 1", second.Size())
}
@@ -386,14 +294,44 @@ func TestPersistsRoomReplacementAsOneAuthorityTransition(t *testing.T) {
}
}
func TestRollsBackReplacementWhenOldAuthorityIsStale(t *testing.T) {
path := databasePath(t)
first := stableStore(t, path, Options{})
createRoom(t, first, protocol.CodeEntryPrivate, "room-password", "4321")
if err := first.Close(); err != nil {
t.Fatal(err)
}
database, err := NewDatabase(path)
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = database.Close() })
before, err := database.Initialize()
if err != nil || len(before) != 1 {
t.Fatalf("Initialize = %v, %v", before, err)
}
replacement := before[0]
replacement.RoomID = "5678"
err = database.ReplaceRoom("4321", make([]byte, 32), replacement)
if err == nil || err.Error() != "Room database replacement delete did not match current authority" {
t.Fatalf("ReplaceRoom = %v", err)
}
if err := database.Close(); err != nil {
t.Fatal(err)
}
after, err := database.Initialize()
if err != nil || !reflect.DeepEqual(after, before) {
t.Fatalf("failed replacement changed durable authority: %v", err)
}
}
// TestOpensThePathWithURIReservedCharactersLiterally: `new DatabaseSync(path)`
// opened the literal path, while the driver's DSN is a URI in which `%41`
// decodes to `A` and `#` ends the file name; the file must still be the
// configured one, and the busy-timeout pragma must survive the escaping.
func TestOpensThePathWithURIReservedCharactersLiterally(t *testing.T) {
testClock := &clock{nowMs: 10}
path := filepath.Join(t.TempDir(), "rooms%41#1.sqlite")
first := stableStore(t, path, testClock, Options{})
first := stableStore(t, path, Options{})
createRoom(t, first, protocol.CodeEntryOpen, "", "4321")
if err := first.Close(); err != nil {
t.Fatalf("Close: %v", err)
@@ -401,7 +339,7 @@ func TestOpensThePathWithURIReservedCharactersLiterally(t *testing.T) {
if _, err := os.Stat(path); err != nil {
t.Fatalf("the configured file was not the one opened: %v", err)
}
if second := stableStore(t, path, testClock, Options{}); second.Size() != 1 {
if second := stableStore(t, path, Options{}); second.Size() != 1 {
t.Fatalf("restored size = %d, want 1", second.Size())
}
}
@@ -412,14 +350,14 @@ func TestRejectsAnotherOwnerOfTheSameDatabase(t *testing.T) {
if err != nil {
t.Fatalf("NewDatabase: %v", err)
}
if _, err := first.Initialize(0, 1_000); err != nil {
if _, err := first.Initialize(); err != nil {
t.Fatalf("Initialize: %v", err)
}
contender, err := NewDatabase(path)
if err != nil {
t.Fatalf("NewDatabase: %v", err)
}
_, err = contender.Initialize(0, 1_000)
_, err = contender.Initialize()
if err == nil || !strings.Contains(strings.ToLower(err.Error()), "locked") {
t.Fatalf("contender Initialize = %v, want a lock failure", err)
}
@@ -434,7 +372,7 @@ func TestRejectsAnotherOwnerOfTheSameDatabase(t *testing.T) {
if err != nil {
t.Fatalf("NewDatabase: %v", err)
}
restored, err := successor.Initialize(0, 1_000)
restored, err := successor.Initialize()
if err != nil || len(restored) != 0 {
t.Fatalf("successor Initialize = %v, %v", restored, err)
}
@@ -446,7 +384,7 @@ func TestRejectsAnotherOwnerOfTheSameDatabase(t *testing.T) {
func TestRejectsAnUnknownDatabaseIdentity(t *testing.T) {
for _, testCase := range []struct{ name, mutation string }{
{"application identity", "PRAGMA application_id = 1"},
{"schema version", "PRAGMA user_version = 2"},
{"schema version", "PRAGMA user_version = 99"},
} {
t.Run(testCase.name, func(t *testing.T) {
path := databasePath(t)
@@ -454,7 +392,7 @@ func TestRejectsAnUnknownDatabaseIdentity(t *testing.T) {
if err != nil {
t.Fatalf("NewDatabase: %v", err)
}
if _, err := initial.Initialize(0, 1_000); err != nil {
if _, err := initial.Initialize(); err != nil {
t.Fatalf("Initialize: %v", err)
}
if err := initial.Close(); err != nil {
@@ -466,7 +404,7 @@ func TestRejectsAnUnknownDatabaseIdentity(t *testing.T) {
if err != nil {
t.Fatalf("NewDatabase: %v", err)
}
_, err = reopened.Initialize(0, 1_000)
_, err = reopened.Initialize()
if err == nil || !strings.Contains(err.Error(), "does not match") {
t.Fatalf("Initialize = %v", err)
}
@@ -483,7 +421,7 @@ func TestRejectsAnAlteredCurrentVersionSchema(t *testing.T) {
if err != nil {
t.Fatalf("NewDatabase: %v", err)
}
if _, err := initial.Initialize(0, 1_000); err != nil {
if _, err := initial.Initialize(); err != nil {
t.Fatalf("Initialize: %v", err)
}
if err := initial.Close(); err != nil {
@@ -495,7 +433,7 @@ func TestRejectsAnAlteredCurrentVersionSchema(t *testing.T) {
if err != nil {
t.Fatalf("NewDatabase: %v", err)
}
_, err = reopened.Initialize(0, 1_000)
_, err = reopened.Initialize()
if err == nil || err.Error() != "Room database columns do not match the current schema" {
t.Fatalf("Initialize = %v", err)
}
@@ -506,8 +444,7 @@ func TestRejectsAnAlteredCurrentVersionSchema(t *testing.T) {
func TestRejectsCorruptAuthorityRows(t *testing.T) {
path := databasePath(t)
testClock := &clock{nowMs: 100}
initial := stableStore(t, path, testClock, Options{})
initial := stableStore(t, path, Options{})
createRoom(t, initial, protocol.CodeEntryOpen, "", "7890")
if err := initial.Close(); err != nil {
t.Fatalf("Close: %v", err)
@@ -529,10 +466,8 @@ func TestRejectsCorruptAuthorityRows(t *testing.T) {
t.Fatalf("NewDatabase: %v", err)
}
reopened, err := New(Options{
LeaseMs: 1_000,
MaxRooms: 8,
MaxViewersPerRoom: 8,
Now: func() int64 { return testClock.nowMs },
Database: database,
})
if err != nil {
@@ -550,9 +485,7 @@ func TestRejectsCorruptAuthorityRows(t *testing.T) {
}
}
// TestRejectsInvalidAuthorityArguments keeps the assertions the TypeScript ran
// before touching SQLite: STRICT would accept an integral REAL or a numeric
// TEXT timestamp, and a caller mistake must not reach the file at all.
// Invalid authority must be rejected before mutating SQLite.
func TestRejectsInvalidAuthorityArguments(t *testing.T) {
valid := StoredRoomAuthority{
RoomID: "1234",
@@ -561,7 +494,6 @@ func TestRejectsInvalidAuthorityArguments(t *testing.T) {
ViewerAuthorizationGeneration: "MzMzMzMzMzMzMzMzMzMzMw",
CodeEntryPolicy: protocol.CodeEntryOpen,
}
lease := int64(protocol.MaxSafeInteger + 1)
shortDigest := make([]byte, 31)
for _, testCase := range []struct {
name string
@@ -598,13 +530,8 @@ func TestRejectsInvalidAuthorityArguments(t *testing.T) {
room.ViewerPasswordMaterial = make([]byte, 47)
return d.InsertRoom(room)
}, "Viewer password material must contain 48 bytes"},
{"lease deadline", func(d *Database) error {
room := valid
room.LeaseExpiresAtMs = &lease
return d.InsertRoom(room)
}, "Room database lease deadline must be a safe positive integer"},
{"identity", func(d *Database) error {
return d.SetLeaseDeadline("123", make([]byte, 32), nil)
return d.DeleteRoom("123", make([]byte, 32))
}, "Room ID is invalid"},
{"update policy", func(d *Database) error {
return d.SetCodeEntryPolicy("1234", make([]byte, 32), "public")
@@ -618,7 +545,7 @@ func TestRejectsInvalidAuthorityArguments(t *testing.T) {
if err != nil {
t.Fatalf("NewDatabase: %v", err)
}
if _, err := database.Initialize(0, 1_000); err != nil {
if _, err := database.Initialize(); err != nil {
t.Fatalf("Initialize: %v", err)
}
defer func() { _ = database.Close() }()
@@ -644,10 +571,10 @@ func TestRejectsASecondInitialize(t *testing.T) {
if err != nil {
t.Fatalf("NewDatabase: %v", err)
}
if _, err := database.Initialize(0, 1_000); err != nil {
if _, err := database.Initialize(); err != nil {
t.Fatalf("Initialize: %v", err)
}
if _, err := database.Initialize(0, 1_000); err == nil ||
if _, err := database.Initialize(); err == nil ||
err.Error() != "Room database is already initialized" {
t.Fatalf("second Initialize = %v", err)
}
@@ -664,21 +591,3 @@ func TestRejectsASecondInitialize(t *testing.T) {
t.Fatalf("InsertRoom after Close = %v", err)
}
}
func TestRejectsAnActiveLeaseThatIsNotInTheFuture(t *testing.T) {
database, err := NewDatabase(databasePath(t))
if err != nil {
t.Fatalf("NewDatabase: %v", err)
}
if _, err := database.Initialize(1_000, 1_000); err == nil ||
err.Error() != "Room database active-room lease deadline must be in the future" {
t.Fatalf("Initialize = %v", err)
}
if _, err := database.Initialize(-1, 1_000); err == nil ||
err.Error() != "Room database startup time must be a safe positive integer" {
t.Fatalf("Initialize = %v", err)
}
if err := database.Close(); err != nil {
t.Fatalf("Close: %v", err)
}
}
internal/server/room/doc.go
+6 -7
View File
@@ -1,8 +1,8 @@
// Package room ports the room authority: src/server/room-store.ts (rooms,
// participants, room codes, viewer credentials, the four-digit code pool) and
// src/server/room-database.ts (the SQLite stable authority). Every mutation
// writes SQLite before memory and returns the write error unchanged; there is
// no catch-and-continue anywhere, exactly as in the TypeScript.
// src/server/room-database.ts (the SQLite stable authority). Authority mutations
// write SQLite before memory and return the write error unchanged. Participants
// remain process-only.
//
// # Locking
//
@@ -21,8 +21,7 @@
// POST /api/rooms (createRoom):
//
// mu.Lock()
// lease, err := store.BeginCreateRoom() // ROOM_LIMIT + lease deadline, pre-KDF
// if err != nil { mu.Unlock(); return err }
// if err := store.BeginCreateRoom(); err != nil { mu.Unlock(); return err }
// var material []byte
// var derived error
// if password != "" {
@@ -30,7 +29,7 @@
// material, derived = store.DeriveViewerPasswordMaterial(password, nil)
// mu.Lock()
// }
// created, err := store.CreateRoom(policy, material, derived, preferredRoomID, lease)
// created, err := store.CreateRoom(policy, material, derived, preferredRoomID)
// mu.Unlock()
//
// replaceRoom and set-viewer-password (the mayStart closure is the TypeScript
@@ -38,7 +37,7 @@
// the derive goroutine, so it takes the lock itself):
//
// mu.Lock()
// current, err := store.HostManagedRoom(roomID, hostToken) // may fail INVALID_TOKEN/ROOM_EXPIRED
// current, err := store.HostManagedRoom(roomID, hostToken) // may fail INVALID_TOKEN
// if err != nil { mu.Unlock(); return err }
// var material []byte
// var derived error
internal/server/room/store.go
+15 -126
View File
@@ -12,7 +12,6 @@ import (
"math"
"slices"
"strconv"
"time"
"github.com/TNTcraftHIM/Screener/internal/server/ordered"
"github.com/TNTcraftHIM/Screener/internal/server/protocol"
@@ -32,7 +31,6 @@ const (
CodeRoomAccessDenied ErrorCode = "ROOM_ACCESS_DENIED"
CodeRoomBusy ErrorCode = "ROOM_BUSY"
CodeRoomNotFound ErrorCode = "ROOM_NOT_FOUND"
CodeRoomExpired ErrorCode = "ROOM_EXPIRED"
CodeRoomFull ErrorCode = "ROOM_FULL"
CodeHostAlreadyConnected ErrorCode = "HOST_ALREADY_CONNECTED"
CodeRoomLimit ErrorCode = "ROOM_LIMIT"
@@ -78,8 +76,6 @@ type Room struct {
viewerPasswordMaterial []byte
viewerAuthorizationGeneration string
codeEntryPolicy protocol.CodeEntryPolicy
sharingActive bool
leaseExpiresAtMs *int64
host *participant
// ordering: JS Map insertion order decides viewer presence order, the
// closed-session list and the router's participant order.
@@ -92,7 +88,6 @@ type CreatedRoom struct {
HostToken string
CodeEntryPolicy protocol.CodeEntryPolicy
ViewerGrant string
ExpiresAt *string
}
// ConnectParticipantInput is ConnectParticipantInput; Token is read for a host
@@ -121,7 +116,6 @@ type ConnectedParticipant struct {
RoomID string
Role protocol.Role
PeerID string
ExpiresAt *string
HostOnline bool
ReplacedSessionID string
CodeEntryPolicy protocol.CodeEntryPolicy
@@ -177,24 +171,20 @@ type CodeEntryUpdate struct {
// Options is RoomStoreOptions.
type Options struct {
LeaseMs int64
MaxRooms int
MaxViewersPerRoom int
Database *Database
Now func() int64
Random func(size int) []byte
}
// Store is RoomStore. See the package comment for the locking contract.
type Store struct {
// ordering: JS Map insertion order decides AbandonAllRooms/ExpireRooms.
// ordering: JS Map insertion order decides AbandonAllRooms.
rooms ordered.Map[string, *Room]
freeRoomCodes []string
now func() int64
random func(size int) []byte
database *Database
initialized bool
leaseMs int64
maxRooms int
maxViewersPerRoom int
// gate is per store; the TypeScript module-level gate was shared by every
@@ -205,9 +195,6 @@ type Store struct {
// New is the RoomStore constructor; the TypeScript threw where this returns an
// error.
func New(options Options) (*Store, error) {
if options.LeaseMs <= 0 || options.LeaseMs > protocol.MaxSafeInteger {
return nil, errors.New("Room lease must be a positive integer")
}
if options.MaxRooms <= 0 || options.MaxRooms > Capacity {
return nil, fmt.Errorf("Room limit must be an integer between 1 and %d", Capacity)
}
@@ -219,11 +206,9 @@ func New(options Options) (*Store, error) {
}
store := &Store{
freeRoomCodes: make([]string, Capacity),
now: options.Now,
random: options.Random,
database: options.Database,
initialized: options.Database == nil,
leaseMs: options.LeaseMs,
maxRooms: options.MaxRooms,
maxViewersPerRoom: options.MaxViewersPerRoom,
gate: gate{limit: gateActiveLimit, pendingLimit: gatePendingLimit},
@@ -231,9 +216,6 @@ func New(options Options) (*Store, error) {
for index := range store.freeRoomCodes {
store.freeRoomCodes[index] = strconv.Itoa(roomCodeFirst + index)
}
if store.now == nil {
store.now = func() int64 { return time.Now().UnixMilli() }
}
if store.random == nil {
store.random = cryptoRandom
}
@@ -270,7 +252,7 @@ func (s *Store) Initialize() error {
s.initialized = true
return nil
}
if err := s.restore(s.now()); err != nil {
if err := s.restore(); err != nil {
s.rooms.Clear()
_ = s.database.Close()
return err
@@ -279,12 +261,8 @@ func (s *Store) Initialize() error {
return nil
}
func (s *Store) restore(startupNowMs int64) error {
lease, err := s.leaseDeadline(startupNowMs)
if err != nil {
return err
}
storedRooms, err := s.database.Initialize(startupNowMs, lease)
func (s *Store) restore() error {
storedRooms, err := s.database.Initialize()
if err != nil {
return err
}
@@ -306,8 +284,6 @@ func (s *Store) restore(startupNowMs int64) error {
viewerPasswordMaterial: bytes.Clone(stored.ViewerPasswordMaterial),
viewerAuthorizationGeneration: stored.ViewerAuthorizationGeneration,
codeEntryPolicy: stored.CodeEntryPolicy,
sharingActive: false,
leaseExpiresAtMs: stored.LeaseExpiresAtMs,
})
}
available := s.freeRoomCodes[:0]
@@ -320,28 +296,25 @@ func (s *Store) restore(startupNowMs int64) error {
return nil
}
// BeginCreateRoom runs the part of createRoom that precedes the password
// derivation: the room limit and the lease deadline, which the TypeScript took
// from the clock before awaiting the KDF. Pass the deadline to CreateRoom.
func (s *Store) BeginCreateRoom() (leaseExpiresAtMs int64, err error) {
// BeginCreateRoom checks room capacity before password derivation.
func (s *Store) BeginCreateRoom() error {
if err := s.ensureInitialized(); err != nil {
return 0, err
return err
}
if s.rooms.Len() >= s.maxRooms {
return 0, codeError(CodeRoomLimit)
return codeError(CodeRoomLimit)
}
return s.leaseDeadline(s.now())
return nil
}
// CreateRoom is the commit half of createRoom. material and derived are the
// results of DeriveViewerPasswordMaterial (both zero when no password was
// given) and leaseExpiresAtMs comes from BeginCreateRoom.
// given).
func (s *Store) CreateRoom(
policy protocol.CodeEntryPolicy,
material []byte,
derived error,
preferredRoomID string,
leaseExpiresAtMs int64,
) (CreatedRoom, error) {
// createViewerPasswordMaterial threw before any of the checks below.
if derived != nil && !isGateOutcome(derived) {
@@ -363,7 +336,7 @@ func (s *Store) CreateRoom(
if err != nil {
return CreatedRoom{}, err
}
room, created, err := s.newRoom(roomID, policy, material, leaseExpiresAtMs)
room, created, err := s.newRoom(roomID, policy, material)
if err == nil {
err = s.writeDatabase(func() error { return s.database.InsertRoom(storedRoomAuthority(room)) })
}
@@ -437,11 +410,7 @@ func (s *Store) replaceRoom(
material []byte,
current *Room,
) (ReplacedRoom, error) {
leaseExpiresAtMs, err := s.leaseDeadline(s.now())
if err != nil {
return ReplacedRoom{}, err
}
room, created, err := s.newRoom(replacementRoomID, policy, material, leaseExpiresAtMs)
room, created, err := s.newRoom(replacementRoomID, policy, material)
if err != nil {
return ReplacedRoom{}, err
}
@@ -591,7 +560,7 @@ func (s *Store) ConnectParticipant(input ConnectParticipantInput) (ConnectedPart
room, err := s.getAvailableRoom(input.RoomID)
if err != nil {
if input.Role == protocol.RoleViewer && input.ViewerGrant == "" &&
(hasCode(err, CodeInvalidToken) || hasCode(err, CodeRoomExpired)) {
hasCode(err, CodeInvalidToken) {
return ConnectedParticipant{}, codeError(CodeRoomNotFound)
}
return ConnectedParticipant{}, err
@@ -633,8 +602,8 @@ func (s *Store) ViewerGrantMayEnter(roomID, grant string) (bool, error) {
// ViewerPasswordChallenge is the locked first half of connectViewerWithPassword:
// it returns the salt to derive against, a copy of the material to compare with
// and the room reference to re-check afterwards. A missing room, an expired
// room and a room without a private password all answer with the same 48 zero
// and the room reference to re-check afterwards. A missing room and a room
// without a private password both answer with the same 48 zero
// bytes so the derivation cost does not reveal which. The only error is the
// `ensureInitialized()` the TypeScript ran before the password check.
func (s *Store) ViewerPasswordChallenge(
@@ -711,19 +680,7 @@ func (s *Store) DisconnectParticipant(
if room.host.sessionID != sessionID {
return nil, nil
}
leaseExpiresAtMs, err := s.leaseDeadline(s.now())
if err != nil {
return nil, err
}
err = s.writeDatabase(func() error {
return s.database.SetLeaseDeadline(roomID, room.hostTokenDigest, &leaseExpiresAtMs)
})
if err != nil {
return nil, err
}
room.host.sessionID = ""
room.sharingActive = false
room.leaseExpiresAtMs = &leaseExpiresAtMs
return &DisconnectedParticipant{
RoomID: roomID, Role: protocol.RoleHost, PeerID: peerID}, nil
}
@@ -841,35 +798,6 @@ func (s *Store) AbandonAllRooms() ([]ClosedRoom, error) {
return closedRooms, nil
}
// ExpireRooms is expireRooms, in room insertion order. The TypeScript defaulted
// nowMs to the injected clock; every caller passed it explicitly.
func (s *Store) ExpireRooms(nowMs int64) ([]ClosedRoom, error) {
if err := s.ensureInitialized(); err != nil {
return nil, err
}
var expiredIDs []string
var identities []RoomIdentity
var expired []ClosedRoom
for roomID, room := range s.rooms.All() {
if !roomIsExpired(room, nowMs) {
continue
}
expiredIDs = append(expiredIDs, roomID)
identities = append(identities, RoomIdentity{
RoomID: roomID, HostTokenDigest: room.hostTokenDigest})
expired = append(expired, ClosedRoom{
RoomID: roomID, SessionIDs: connectedSessionIDs(room)})
}
if err := s.writeDatabase(func() error { return s.database.DeleteRooms(identities) }); err != nil {
return nil, err
}
for _, roomID := range expiredIDs {
s.rooms.Delete(roomID)
s.releaseRoomCode(roomID)
}
return expired, nil
}
func (s *Store) connectHost(
room *Room, input ConnectParticipantInput,
) (ConnectedParticipant, error) {
@@ -888,16 +816,8 @@ func (s *Store) connectHost(
clientID: input.ClientID, peerID: peerID, admittedBy: admittedByCode}
}
replacedSessionID := joining.sessionID
err := s.writeDatabase(func() error {
return s.database.SetLeaseDeadline(room.roomID, room.hostTokenDigest, nil)
})
if err != nil {
return ConnectedParticipant{}, err
}
joining.sessionID = input.SessionID
room.host = joining
room.sharingActive = true
room.leaseExpiresAtMs = nil
return connectedParticipant(
room, joining.peerID, protocol.RoleHost, replacedSessionID, input.SessionID), nil
@@ -936,7 +856,6 @@ func connectedParticipant(
RoomID: room.roomID,
Role: role,
PeerID: peerID,
ExpiresAt: formatExpiresAt(room.leaseExpiresAtMs),
HostOnline: room.host != nil && room.host.sessionID != "",
ReplacedSessionID: replacedSessionID,
CodeEntryPolicy: room.codeEntryPolicy,
@@ -950,9 +869,6 @@ func (s *Store) getAvailableRoom(roomID string) (*Room, error) {
if !ok {
return nil, codeError(CodeInvalidToken)
}
if roomIsExpired(room, s.now()) {
return nil, codeError(CodeRoomExpired)
}
return room, nil
}
@@ -1067,7 +983,6 @@ func (s *Store) newRoom(
roomID string,
policy protocol.CodeEntryPolicy,
material []byte,
leaseExpiresAtMs int64,
) (*Room, CreatedRoom, error) {
hostTokenBytes := s.random(32)
if len(hostTokenBytes) != 32 {
@@ -1089,15 +1004,12 @@ func (s *Store) newRoom(
viewerPasswordMaterial: material,
viewerAuthorizationGeneration: generation,
codeEntryPolicy: policy,
sharingActive: false,
leaseExpiresAtMs: &leaseExpiresAtMs,
}
return room, CreatedRoom{
RoomID: roomID,
HostToken: hostToken,
CodeEntryPolicy: policy,
ViewerGrant: viewerGrant,
ExpiresAt: formatExpiresAt(&leaseExpiresAtMs),
}, nil
}
@@ -1117,15 +1029,6 @@ func (s *Store) newAuthorizationGeneration() (string, error) {
return base64.RawURLEncoding.EncodeToString(value), nil
}
func (s *Store) leaseDeadline(nowMs int64) (int64, error) {
deadline := nowMs + s.leaseMs
if deadline <= nowMs || deadline > protocol.MaxSafeInteger ||
deadline < -protocol.MaxSafeInteger {
return 0, errors.New("Room lease deadline exceeds the supported time range")
}
return deadline, nil
}
func (s *Store) ensureInitialized() error {
if !s.initialized {
return errors.New("RoomStore stable authority is not initialized")
@@ -1150,23 +1053,9 @@ func storedRoomAuthority(room *Room) StoredRoomAuthority {
ViewerAuthorizationGeneration: room.viewerAuthorizationGeneration,
CodeEntryPolicy: room.codeEntryPolicy,
ViewerPasswordMaterial: room.viewerPasswordMaterial,
LeaseExpiresAtMs: room.leaseExpiresAtMs,
}
}
func roomIsExpired(room *Room, nowMs int64) bool {
return !room.sharingActive && room.leaseExpiresAtMs != nil &&
*room.leaseExpiresAtMs <= nowMs
}
func formatExpiresAt(expiresAtMs *int64) *string {
if expiresAtMs == nil {
return nil
}
value := time.UnixMilli(*expiresAtMs).UTC().Format("2006-01-02T15:04:05.000Z")
return &value
}
func findViewerByPeerID(room *Room, peerID string) *participant {
for _, viewer := range room.viewers.Values() {
if viewer.peerID == peerID {
internal/server/room/store_test.go
+36 -93
View File
@@ -16,21 +16,14 @@ import (
// The scenarios below are ported from tests/room-store.test.ts.
type clock struct{ nowMs int64 }
func newStore(t *testing.T, options Options) (*Store, *clock) {
func newStore(t *testing.T, options Options) *Store {
t.Helper()
testClock := &clock{}
if options.LeaseMs == 0 {
options.LeaseMs = 1_000
}
if options.MaxRooms == 0 {
options.MaxRooms = 2
}
if options.MaxViewersPerRoom == 0 {
options.MaxViewersPerRoom = 2
}
options.Now = func() int64 { return testClock.nowMs }
store, err := New(options)
if err != nil {
t.Fatalf("New: %v", err)
@@ -39,7 +32,7 @@ func newStore(t *testing.T, options Options) (*Store, *clock) {
t.Fatalf("Initialize: %v", err)
}
t.Cleanup(func() { _ = store.Close() })
return store, testClock
return store
}
func constantRandom(value func(size int) byte) func(size int) []byte {
@@ -51,8 +44,7 @@ func constantRandom(value func(size int) byte) func(size int) []byte {
func tryCreateRoom(
store *Store, policy protocol.CodeEntryPolicy, password, preferredRoomID string,
) (CreatedRoom, error) {
lease, err := store.BeginCreateRoom()
if err != nil {
if err := store.BeginCreateRoom(); err != nil {
return CreatedRoom{}, err
}
var material []byte
@@ -60,7 +52,7 @@ func tryCreateRoom(
if password != "" {
material, derived = store.DeriveViewerPasswordMaterial(password, nil)
}
return store.CreateRoom(policy, material, derived, preferredRoomID, lease)
return store.CreateRoom(policy, material, derived, preferredRoomID)
}
func createRoom(
@@ -167,7 +159,7 @@ var roomCodePattern = regexp.MustCompile(`^[1-9][0-9]{3}$`)
var viewerGrantPattern = regexp.MustCompile(`^[A-Za-z0-9_-]{21}[AQgw]$`)
func TestAllocatesEveryFreeRoomCodeAndRecyclesReleases(t *testing.T) {
store, _ := newStore(t, Options{
store := newStore(t, Options{
MaxRooms: Capacity,
Random: constantRandom(func(int) byte { return 0 }),
})
@@ -200,7 +192,7 @@ func TestAllocatesEveryFreeRoomCodeAndRecyclesReleases(t *testing.T) {
}
func TestUsesAFreePreferredCodeAndFallsBack(t *testing.T) {
store, _ := newStore(t, Options{MaxRooms: 3})
store := newStore(t, Options{MaxRooms: 3})
preferred := createRoom(t, store, protocol.CodeEntryOpen, "", "4321")
fallback := createRoom(t, store, protocol.CodeEntryOpen, "", "4321")
@@ -220,7 +212,7 @@ func TestUsesAFreePreferredCodeAndFallsBack(t *testing.T) {
}
func TestAbandonsEveryCurrentRoomAsOneShutdown(t *testing.T) {
store, _ := newStore(t, Options{MaxRooms: 3})
store := newStore(t, Options{MaxRooms: 3})
first := createRoom(t, store, protocol.CodeEntryOpen, "", "")
second := createRoom(t, store, protocol.CodeEntryOpen, "", "")
mustConnect(t, store, hostInput(first.RoomID, first.HostToken))
@@ -248,7 +240,7 @@ func TestAbandonsEveryCurrentRoomAsOneShutdown(t *testing.T) {
}
func TestAtomicallyReplacesARoomWithDifferentAuthority(t *testing.T) {
store, _ := newStore(t, Options{})
store := newStore(t, Options{})
original := createRoom(t, store, protocol.CodeEntryPrivate, "old-password", "4321")
host := mustConnect(t, store, hostInput(original.RoomID, original.HostToken))
mustConnect(t, store, viewerInput(original.RoomID, "viewer-session", original.ViewerGrant))
@@ -293,56 +285,36 @@ func TestAtomicallyReplacesARoomWithDifferentAuthority(t *testing.T) {
}
func TestRejectsRoomLimitsBeyondTheCodeSpace(t *testing.T) {
_, err := New(Options{LeaseMs: 1_000, MaxRooms: Capacity + 1, MaxViewersPerRoom: 2})
_, err := New(Options{MaxRooms: Capacity + 1, MaxViewersPerRoom: 2})
if err == nil || err.Error() != "Room limit must be an integer between 1 and 9000" {
t.Fatalf("New: %v", err)
}
}
func TestKeepsActiveSharingAliveAndRenewsOnlyWithTheExactHostToken(t *testing.T) {
store, testClock := newStore(t, Options{LeaseMs: 1_000})
func TestResumesDisconnectedRoomOnlyWithTheExactHostToken(t *testing.T) {
store := newStore(t, Options{})
room := createRoom(t, store, protocol.CodeEntryOpen, "", "")
host := mustConnect(t, store, hostInput(room.RoomID, room.HostToken))
testClock.nowMs = 2_000
if expired, err := store.ExpireRooms(testClock.nowMs); err != nil || len(expired) != 0 {
t.Fatalf("ExpireRooms while sharing = %v, %v", expired, err)
}
if _, err := store.DisconnectParticipant(room.RoomID, host.PeerID, "host-session"); err != nil {
t.Fatalf("DisconnectParticipant: %v", err)
}
testClock.nowMs = 2_500
if expired, err := store.ExpireRooms(testClock.nowMs); err != nil || len(expired) != 0 {
t.Fatalf("ExpireRooms inside the lease = %v, %v", expired, err)
}
_, err := store.ConnectParticipant(hostInput(room.RoomID, "wrong-token"))
expectCode(t, err, CodeInvalidToken)
resumed := mustConnect(t, store, hostInput(room.RoomID, room.HostToken, "host-session-2"))
if resumed.ExpiresAt != nil {
t.Fatalf("resumed expiresAt = %v, want null", *resumed.ExpiresAt)
}
if _, err := store.DisconnectParticipant(
room.RoomID, resumed.PeerID, "host-session-2"); err != nil {
t.Fatalf("DisconnectParticipant: %v", err)
}
mustConnect(t, store, viewerInput(room.RoomID, "viewer-session", room.ViewerGrant))
testClock.nowMs = 3_501
expired, err := store.ExpireRooms(testClock.nowMs)
if err != nil || len(expired) != 1 {
t.Fatalf("ExpireRooms after the lease = %v, %v", expired, err)
}
_, err = store.ConnectParticipant(hostInput(room.RoomID, room.HostToken, "late-host"))
expectCode(t, err, CodeInvalidToken)
mustConnect(t, store, hostInput(room.RoomID, room.HostToken, "late-host"))
}
func TestManagesDormantAccessWithTheExactHostTokenWithoutRenewing(t *testing.T) {
store, testClock := newStore(t, Options{LeaseMs: 1_000})
func TestManagesDormantAccessWithTheExactHostToken(t *testing.T) {
store := newStore(t, Options{})
room := createRoom(t, store, protocol.CodeEntryOpen, "", "")
otherRoom := createRoom(t, store, protocol.CodeEntryOpen, "", "")
testClock.nowMs = 900
_, err := trySetViewerPassword(store, room.RoomID, "room-password", "wrong-token")
expectCode(t, err, CodeInvalidToken)
@@ -370,28 +342,10 @@ func TestManagesDormantAccessWithTheExactHostTokenWithoutRenewing(t *testing.T)
if _, ok := store.GetConnectedHost(room.RoomID); ok {
t.Fatal("a dormant room reported a connected host")
}
testClock.nowMs = 1_001
expired, err := store.ExpireRooms(testClock.nowMs)
if err != nil {
t.Fatalf("ExpireRooms: %v", err)
}
if !slicesContainsRoom(expired, room.RoomID) {
t.Fatalf("expired = %+v, want %q", expired, room.RoomID)
}
}
func slicesContainsRoom(rooms []ClosedRoom, roomID string) bool {
for _, room := range rooms {
if room.RoomID == roomID {
return true
}
}
return false
}
func TestKeepsViewerGrantsIndependentFromCodeEntryPolicy(t *testing.T) {
store, _ := newStore(t, Options{MaxRooms: 3})
store := newStore(t, Options{MaxRooms: 3})
room := createRoom(t, store, protocol.CodeEntryPrivate, "", "")
_, err := store.ConnectParticipant(viewerInput(room.RoomID, "code-only", ""))
@@ -410,14 +364,13 @@ func TestKeepsViewerGrantsIndependentFromCodeEntryPolicy(t *testing.T) {
}
func TestKeepsTheExactViewerGrantValidForTheRoomIncarnation(t *testing.T) {
store, testClock := newStore(t, Options{LeaseMs: 1_000})
store := newStore(t, Options{})
room := createRoom(t, store, protocol.CodeEntryOpen, "", "")
mustConnect(t, store, hostInput(room.RoomID, room.HostToken))
if !viewerGrantPattern.MatchString(room.ViewerGrant) {
t.Fatalf("viewer grant = %q", room.ViewerGrant)
}
testClock.nowMs = 8 * 24 * 60 * 60 * 1_000
connected := mustConnect(t, store, viewerInput(room.RoomID, "room-lived-grant", room.ViewerGrant))
if connected.Role != protocol.RoleViewer {
t.Fatalf("role = %q", connected.Role)
@@ -425,7 +378,7 @@ func TestKeepsTheExactViewerGrantValidForTheRoomIncarnation(t *testing.T) {
}
func TestSupportsOpenAndPasswordEnabledPrivateCodeEntry(t *testing.T) {
store, _ := newStore(t, Options{MaxRooms: 3})
store := newStore(t, Options{MaxRooms: 3})
open := createRoom(t, store, protocol.CodeEntryOpen, "", "")
if connected := mustConnect(t, store, viewerInput(open.RoomID, "open", "")); connected.Role !=
protocol.RoleViewer {
@@ -495,7 +448,7 @@ func waitForSaturatedGate(t *testing.T, store *Store) {
}
func TestDoesNotCreateAPasswordlessRoomWhenTheGateIsBusy(t *testing.T) {
store, _ := newStore(t, Options{MaxRooms: 1})
store := newStore(t, Options{MaxRooms: 1})
withSaturatedGate(t, store, func() {
_, err := tryCreateRoom(store, protocol.CodeEntryPrivate, "room-password", "4321")
@@ -510,7 +463,7 @@ func TestDoesNotCreateAPasswordlessRoomWhenTheGateIsBusy(t *testing.T) {
}
func TestKeepsGateSaturationDistinctFromInvalidCredentials(t *testing.T) {
store, _ := newStore(t, Options{})
store := newStore(t, Options{})
room := createRoom(t, store, protocol.CodeEntryPrivate, "room-password", "")
withSaturatedGate(t, store, func() {
@@ -525,7 +478,7 @@ func TestKeepsGateSaturationDistinctFromInvalidCredentials(t *testing.T) {
}
func TestKeepsRoomReplacementAndPasswordUpdatesUnchangedWhenBusy(t *testing.T) {
store, _ := newStore(t, Options{MaxRooms: 3})
store := newStore(t, Options{MaxRooms: 3})
room := createRoom(t, store, protocol.CodeEntryPrivate, "old-password", "4321")
withSaturatedGate(t, store, func() {
@@ -560,7 +513,7 @@ func TestKeepsRoomReplacementAndPasswordUpdatesUnchangedWhenBusy(t *testing.T) {
}
func TestBoundsPasswordDerivationsAndUsesTheSamePathForUnknownRooms(t *testing.T) {
store, _ := newStore(t, Options{})
store := newStore(t, Options{})
var mayStartCalls atomic.Int64
release := make(chan struct{})
@@ -611,7 +564,7 @@ func TestBoundsPasswordDerivationsAndUsesTheSamePathForUnknownRooms(t *testing.T
}
func TestRotatesAndRevokesViewerGrantsWithoutChangingCodeEntry(t *testing.T) {
store, _ := newStore(t, Options{})
store := newStore(t, Options{})
room := createRoom(t, store, protocol.CodeEntryOpen, "", "")
mustConnect(t, store, hostInput(room.RoomID, room.HostToken))
@@ -646,7 +599,7 @@ func TestKeepsTheOldGrantWhenRotationCannotCreateAGeneration(t *testing.T) {
failedRandomCall := 0
randomCalls := 0
randomValue := byte(0)
store, _ := newStore(t, Options{
store := newStore(t, Options{
Random: func(size int) []byte {
randomCalls++
randomValue++
@@ -680,9 +633,9 @@ func TestKeepsTheOldGrantWhenRotationCannotCreateAGeneration(t *testing.T) {
}
func TestClearsEveryCredentialOnProcessRestart(t *testing.T) {
firstStore, _ := newStore(t, Options{})
firstStore := newStore(t, Options{})
room := createRoom(t, firstStore, protocol.CodeEntryPrivate, "room-password", "")
secondStore, _ := newStore(t, Options{})
secondStore := newStore(t, Options{})
replacement := createRoom(t, secondStore, protocol.CodeEntryOpen, "", "")
if !roomCodePattern.MatchString(replacement.RoomID) {
@@ -702,7 +655,7 @@ func TestClearsEveryCredentialOnProcessRestart(t *testing.T) {
}
func TestRejectsAnOldGrantWhenANewRoomReusesTheSameCode(t *testing.T) {
firstStore, _ := newStore(t, Options{
firstStore := newStore(t, Options{
Random: constantRandom(func(size int) byte {
if size == 8 {
return 0
@@ -715,7 +668,7 @@ func TestRejectsAnOldGrantWhenANewRoomReusesTheSameCode(t *testing.T) {
t.Fatalf("AbandonRoom: %v", err)
}
secondStore, _ := newStore(t, Options{
secondStore := newStore(t, Options{
Random: constantRandom(func(size int) byte {
if size == 8 {
return 0
@@ -737,7 +690,7 @@ func TestRejectsAnOldGrantWhenANewRoomReusesTheSameCode(t *testing.T) {
}
func TestParticipantReplacementRules(t *testing.T) {
store, _ := newStore(t, Options{})
store := newStore(t, Options{})
room := createRoom(t, store, protocol.CodeEntryOpen, "", "")
host := mustConnect(t, store, hostInput(room.RoomID, room.HostToken))
@@ -823,7 +776,7 @@ func TestHostChangeChecksIdentityAfterTheDerivation(t *testing.T) {
// A constant random source gives a recreated room the same Host token, so
// the identity comparison is reachable instead of being shadowed by the
// token check.
store, testClock := newStore(t, Options{
store := newStore(t, Options{
MaxRooms: 3,
Random: constantRandom(func(int) byte { return 7 }),
})
@@ -856,15 +809,6 @@ func TestHostChangeChecksIdentityAfterTheDerivation(t *testing.T) {
if err != nil {
t.Fatalf("HostManagedRoom: %v", err)
}
testClock.nowMs = 5_000
_, err = store.SetViewerPassword(again.RoomID, again.HostToken, nil, nil, incarnation)
expectCode(t, err, CodeRoomExpired)
// An expired room outranks a saturated gate, as the TypeScript order did.
_, err = store.SetViewerPassword(
again.RoomID, again.HostToken, nil, codeError(CodeRoomBusy), incarnation)
expectCode(t, err, CodeRoomExpired)
testClock.nowMs = 0
_, err = store.SetViewerPassword(
again.RoomID, again.HostToken, nil, errDerivationCancelled, incarnation)
expectCode(t, err, CodeRoomAccessDenied)
@@ -883,7 +827,7 @@ func TestHostChangeChecksIdentityAfterTheDerivation(t *testing.T) {
func TestPasswordLoginRechecksTheRoomAfterTheDerivation(t *testing.T) {
// A constant random source repeats the salt, so a recreated room holds the
// same material and only the room identity separates the two incarnations.
store, _ := newStore(t, Options{
store := newStore(t, Options{
MaxRooms: 3,
Random: constantRandom(func(int) byte { return 7 }),
})
@@ -938,7 +882,7 @@ func TestPasswordLoginRequiresAnInitializedStore(t *testing.T) {
if err != nil {
t.Fatalf("NewDatabase: %v", err)
}
store, err := New(Options{LeaseMs: 1_000, MaxRooms: 2, MaxViewersPerRoom: 2, Database: database})
store, err := New(Options{MaxRooms: 2, MaxViewersPerRoom: 2, Database: database})
if err != nil {
t.Fatalf("New: %v", err)
}
@@ -951,7 +895,7 @@ func TestPasswordLoginRequiresAnInitializedStore(t *testing.T) {
// TestPasswordLoginHidesAFullRoom keeps ROOM_FULL from leaking to a Viewer that
// authenticated with the room password.
func TestPasswordLoginHidesAFullRoom(t *testing.T) {
store, _ := newStore(t, Options{MaxViewersPerRoom: 2})
store := newStore(t, Options{MaxViewersPerRoom: 2})
room := createRoom(t, store, protocol.CodeEntryPrivate, "room-password", "")
mustConnect(t, store, viewerInput(room.RoomID, "first", room.ViewerGrant))
mustConnect(t, store, viewerInput(room.RoomID, "second", room.ViewerGrant))
@@ -966,9 +910,8 @@ func TestPasswordLoginHidesAFullRoom(t *testing.T) {
}
func TestCreateRoomReportsACancelledDerivation(t *testing.T) {
store, _ := newStore(t, Options{})
lease, err := store.BeginCreateRoom()
if err != nil {
store := newStore(t, Options{})
if err := store.BeginCreateRoom(); err != nil {
t.Fatalf("BeginCreateRoom: %v", err)
}
material, derived := store.DeriveViewerPasswordMaterial(
@@ -976,7 +919,7 @@ func TestCreateRoomReportsACancelledDerivation(t *testing.T) {
if material != nil || !errors.Is(derived, errDerivationCancelled) {
t.Fatalf("derive = %v, %v", material, derived)
}
_, err = store.CreateRoom(protocol.CodeEntryPrivate, material, derived, "", lease)
_, err := store.CreateRoom(protocol.CodeEntryPrivate, material, derived, "")
if err == nil || err.Error() != "Room creation password derivation was cancelled" {
t.Fatalf("CreateRoom = %v", err)
}
@@ -986,7 +929,7 @@ func TestCreateRoomReportsACancelledDerivation(t *testing.T) {
}
func TestRejectsAMalformedViewerPassword(t *testing.T) {
store, _ := newStore(t, Options{})
store := newStore(t, Options{})
for _, password := range []string{"", "with space", strings.Repeat("a", 65)} {
material, derived := store.DeriveViewerPasswordMaterial(password, nil)
if material != nil || !hasCode(derived, CodeInvalidToken) {
internal/server/room/testdata/node-rooms.db
Vendored
BIN
View File
Binary file not shown.
internal/server/route/api.go
+2 -2
View File
@@ -306,8 +306,8 @@ type ReconcileResult struct {
Released []*Resource
}
// SettleResult is the outcome of a candidate settlement call. Committed is
// true only when CandidateReady committed the attempt (TS: optional
// SettleResult is the outcome of candidate settlement or active replacement.
// Committed is true only when CandidateReady committed the attempt (TS: optional
// committed, read as truthy by the router). TS: SettleResult.
type SettleResult struct {
Accepted bool
internal/server/route/controller.go
+18 -9
View File
@@ -778,13 +778,13 @@ func (c *Controller) InvalidateHostPublication(guard PublicationGuard, nowMs *in
return true
}
// AdoptDirectConnection rewrites the connection ID of a usable, active
// direct edge the operation does not own; true when adopted (no-op when
// unchanged).
func (c *Controller) AdoptDirectConnection(input AdoptDirectConnectionInput) bool {
// AdoptDirectConnection accepts a replacement for the exact active edge.
// Recovery supersedes optional preparation for that child when the replacement
// arrives. Required availability work, including capacity reduction, keeps priority.
func (c *Controller) AdoptDirectConnection(input AdoptDirectConnectionInput, nowMs int64) SettleResult {
result := SettleResult{ActiveRevision: c.revision}
edge, _ := c.upstreamByViewer.Get(input.ChildPeerID)
if (c.operation != nil && c.operation.childPeerID == input.ChildPeerID) ||
c.revision != input.RouteRevision ||
if c.revision != input.RouteRevision ||
edge == nil ||
edge.Kind != UpstreamPeer ||
edge.Transport != TransportDirect ||
@@ -794,10 +794,17 @@ func (c *Controller) AdoptDirectConnection(input AdoptDirectConnectionInput) boo
edge.ParentSessionID != input.ParentSessionID ||
edge.ConnectionID != input.ConnectionID ||
input.NewConnectionID == "" {
return false
return result
}
if input.NewConnectionID == edge.ConnectionID {
return true
result.Accepted = true
return result
}
if op := c.operation; op != nil && op.childPeerID == input.ChildPeerID {
if isAvailabilityOperation(op.reason) {
return result
}
result.Released = c.abortOperation(&nowMs, RejectionAborted)
}
c.clearQualityForParticipant(input.ChildPeerID)
edge.ConnectionID = input.NewConnectionID
@@ -805,7 +812,9 @@ func (c *Controller) AdoptDirectConnection(input AdoptDirectConnectionInput) boo
child.availabilityExhausted = false
}
c.touchFacts()
return true
result.Accepted = true
result.ActiveRevision = c.revision
return result
}
// RetireHostPublication retires the physically active publication and its
internal/server/route/controller_test.go
+30 -1
View File
@@ -1127,7 +1127,7 @@ func TestClearsQualityShadowForRelaySubtreeWhenSourceChanges(t *testing.T) {
eq(t, routes.AdoptDirectConnection(AdoptDirectConnectionInput{
EdgeGuard: edgeGuard(A, "host_session", revision, "a_from_host"),
NewConnectionID: "a_from_host_recovered",
}), true)
}, 1_002).Accepted, true)
eq(t, withQuality(1_002), 0)
for _, step := range []struct {
@@ -1142,6 +1142,35 @@ func TestClearsQualityShadowForRelaySubtreeWhenSourceChanges(t *testing.T) {
eq(t, withQuality(3_101), 2)
}
func TestActiveRebuildRetiresBootstrapReservationWithoutLosingWaitingDemand(t *testing.T) {
routes := newController(2, Options{SfuEnabled: true})
for _, child := range []string{B, C} {
addViewer(routes, child, 0, nil)
routes.hydrateEdge(child, peerEdge(HOST, child+"_from_host"))
}
addViewer(routes, A, 0, ms(0))
bootstrap := must(t, routes.Reconcile(0).Operation)
eq(t, bootstrap.Reason, DemandSfuBootstrap)
retained := edgeOf(t, routes, bootstrap.ChildPeerID)
attempt := beginOperation(t, routes, BeginInput{
NowMs: 1, ConnectionID: "carrier_sfu", PublicationGeneration: "publication",
PublicationConnectionID: "publication_connection",
Reservation: sfuCreateOverlap("subscription", "publication", "overlap"),
})
replaced := routes.AdoptDirectConnection(AdoptDirectConnectionInput{
EdgeGuard: edgeGuard(bootstrap.ChildPeerID, "host_session", routes.Revision(), retained.ConnectionID),
NewConnectionID: "recovered_connection",
}, 2)
eq(t, replaced.Accepted, true)
eqLabels(t, replaced.Released, "subscription", "publication", "overlap")
noOperation(t, routes.Snapshot().Operation)
eq(t, edgeOf(t, routes, bootstrap.ChildPeerID).ConnectionID, "recovered_connection")
eq(t, routes.CandidateReady(guardFor(bootstrap.ChildPeerID, attempt.Current.Revision, "carrier_sfu"), 3, nil, CandidateProof{}).Accepted, false)
next := must(t, routes.Reconcile(4).Operation)
eq(t, next.Reason, DemandSfuBootstrap)
eq(t, next.DemandPeerID, A)
}
// TS 2569: retains one latest timing sample for a 20-Viewer burst
func TestRetainsOneLatestTimingSampleForTwentyViewerBurst(t *testing.T) {
routes := newController(2, Options{})
internal/server/signal/presence.go
+7 -25
View File
@@ -2,7 +2,7 @@ package signal
// Viewer presence, host status, disconnect handling and room termination of
// src/server/signaling.ts (sendViewerPresence 1594, broadcastHostStatus
// 1683, handleDisconnect 1397, stopSharing 1466, terminateRoom 1500,
// 1683, handleDisconnect 1397, stopSharing 1466, closeRoom,
// revokeGrantViewers 1016, closeRevokedViewerSession 1697 and the
// viewer-keyed map helpers). Every function here runs with mu held.
@@ -123,8 +123,7 @@ func (s *Server) broadcastHostStatus(roomID string, online, paused bool) {
}
// handleDisconnect ports handleDisconnect. It runs once, from the reader
// goroutine's exit (D6). The host lease write is one of the two paths the TS
// left uncaught, so its failure panics (D7).
// goroutine's exit (D6).
func (s *Server) handleDisconnect(sess *session) {
if !s.sessions.Has(sess) {
return
@@ -149,7 +148,7 @@ func (s *Server) handleDisconnect(sess *session) {
disconnected, err := s.store.DisconnectParticipant(
sess.authenticated.roomID, sess.authenticated.peerID, sess.sessionID)
if err != nil {
panic(fmt.Errorf("room store write failed during host disconnect: %w", err))
panic(fmt.Errorf("room store failed during disconnect: %w", err))
}
if disconnected == nil {
return
@@ -215,26 +214,9 @@ func (s *Server) abandonRoom(roomID string) error {
return nil
}
// closeRoom ports closeRoom.
// closeRoom closes the host session first, then
// viewers in store order, each told room-closed then closed with 1000.
func (s *Server) closeRoom(closed room.ClosedRoom) {
s.terminateRoom(closed, "host-ended", "Room abandoned")
}
// expireRooms ports expireRooms, the cleanup timer body. Its store write is
// the second path the TS left uncaught (D7).
func (s *Server) expireRooms() {
expired, err := s.store.ExpireRooms(s.now())
if err != nil {
panic(fmt.Errorf("room store write failed while expiring rooms: %w", err))
}
for _, closed := range expired {
s.terminateRoom(closed, "expired", "Room expired")
}
}
// terminateRoom ports terminateRoom (O4: the host session first, then
// viewers in store order, each told room-closed then closed with 1000).
func (s *Server) terminateRoom(closed room.ClosedRoom, reason, closeText string) {
s.clearRoomGraceTimers(closed.RoomID)
s.clearRoomViewerEvidence(closed.RoomID)
s.router.deleteRoom(closed.RoomID)
@@ -244,8 +226,8 @@ func (s *Server) terminateRoom(closed room.ClosedRoom, reason, closeText string)
if sess == nil {
continue
}
s.send(sess, protocol.RoomClosedMessage{Type: "room-closed", Reason: reason})
sess.close(websocket.StatusNormalClosure, closeText)
s.send(sess, protocol.RoomClosedMessage{Type: "room-closed", Reason: "host-ended"})
sess.close(websocket.StatusNormalClosure, "Room abandoned")
}
}
internal/server/signal/recovery_test.go
+100
View File
@@ -0,0 +1,100 @@
package signal
import (
"strings"
"testing"
"github.com/TNTcraftHIM/Screener/internal/server/protocol"
)
func TestSignalRebuildSupersedesOverlappingQualityPreparation(t *testing.T) {
for _, order := range []string{"prepare-before-restart", "restart-before-prepare"} {
t.Run(order, func(t *testing.T) {
h := startHarness(t, harnessOptions{endpointMediaCopyCapacity: 2})
host := openClient(t, h)
hostAuth := authenticate(t, host, h.room, protocol.RoleHost, "recovery-host", 1,
testShareGeneration, presenceOptions{routePolicy: qualityPolicy})
viewer := openClient(t, h)
viewerAuth := authenticate(t, viewer, h.room, protocol.RoleViewer, "recovery-viewer", 0, "", presenceOptions{})
commitPreparedRoute(t, viewer)
var active activeViewerMediaEdge
var hostSession, viewerSession string
h.locked(func() {
parent, _ := h.store.GetConnectedHost(h.room.RoomID)
child, _ := h.store.GetConnectedViewer(h.room.RoomID, viewerAuth.PeerID)
hostSession, viewerSession = parent.SessionID, child.SessionID
active, _ = h.server.router.resolveActiveViewerMediaEdge(h.room.RoomID, viewerAuth.PeerID)
})
restart := func() {
viewer.sendJSON(map[string]any{"type": "restart-request", "targetPeerId": hostAuth.PeerID,
"connectionId": active.connectionID, "rebuild": true})
expectMatch(t, host.next("restart-request").raw, `{"rebuild":true}`)
}
if order == "restart-before-prepare" {
restart()
}
h.locked(func() {
participant := authenticatedRouteParticipant{roomID: h.room.RoomID, role: protocol.RoleHost,
peerID: hostAuth.PeerID, sessionID: hostSession, routePolicy: qualityPolicy}
for index, state := range []string{"healthy", "degraded", "degraded", "degraded"} {
h.server.router.observeSenderQualityEvidence(participant, senderEvidenceMessage(
viewerAuth.PeerID, active.connectionID, "recovery-sender", "recovery-track",
float64(100+index), active.revision, state))
}
})
pending := nextPreparedRoute(t, viewer)
if !pending.Candidate.QualityProbe || pending.Assignment.Upstream != protocol.PeerUpstream(hostAuth.PeerID) {
t.Fatal("expected a same-parent quality preparation")
}
if !strings.HasPrefix(active.connectionID, candidateConnectionIDPrefix) ||
!strings.HasPrefix(pending.Candidate.ConnectionID, candidateConnectionIDPrefix) {
t.Fatal("prepared connections must remain distinguishable after retirement")
}
if order == "prepare-before-restart" {
restart()
}
// A same-parent probe must not intercept the retained connection's
// offer, answer or ICE. The probe's own signaling also stays valid.
host.sendJSON(signalDescription(viewerAuth.PeerID, active.connectionID, "offer"))
expectMatch(t, viewer.next("signal").raw, `{"payload":{"connectionId":"`+active.connectionID+`"}}`)
viewer.sendJSON(signalDescription(hostAuth.PeerID, active.connectionID, "answer"))
host.next("signal")
viewer.sendJSON(signalCandidate(hostAuth.PeerID, active.connectionID))
host.next("signal")
host.sendJSON(signalDescription(viewerAuth.PeerID, pending.Candidate.ConnectionID, "offer"))
expectMatch(t, viewer.next("signal").raw, `{"payload":{"connectionId":"`+pending.Candidate.ConnectionID+`"}}`)
const replacement = "manual_replacement_connection"
host.sendJSON(signalDescription(viewerAuth.PeerID, replacement, "offer"))
expectMatch(t, viewer.next("signal").raw, `{"payload":{"connectionId":"`+replacement+`"}}`)
settled := nextActiveRouteAfter(t, viewer, pending.Revision)
if settled.Assignment.Upstream != active.upstream {
t.Fatal("manual recovery must keep the committed parent")
}
viewer.sendJSON(signalDescription(hostAuth.PeerID, replacement, "answer"))
host.next("signal")
h.locked(func() {
rm, _ := h.server.router.rooms.Get(h.room.RoomID)
if rm.controller.Snapshot().Operation != nil {
t.Fatal("superseded preparation must release the serial operation")
}
for _, retired := range []string{pending.Candidate.ConnectionID, active.connectionID} {
authorization := h.server.router.peerSignalAuthorization(peerSignalInput{
roomID: h.room.RoomID, sourcePeerID: hostAuth.PeerID, sourceSessionID: hostSession,
targetPeerID: viewerAuth.PeerID, targetSessionID: viewerSession,
connectionID: retired, signalKind: "description", descriptionType: "offer",
})
if authorization != signalAuthorizationDenied {
t.Fatal("a late retired offer must not replace the recovered connection")
}
}
current, _ := h.server.router.resolveActiveViewerMediaEdge(h.room.RoomID, viewerAuth.PeerID)
if current.connectionID != replacement || current.revision != settled.Revision {
t.Fatalf("recovered edge = %+v", current)
}
})
})
}
}
internal/server/signal/router.go
+17 -5
View File
@@ -569,7 +569,12 @@ func (r *router) getViewerRouteUpstream(roomID, viewerPeerID string) protocol.Me
return assignmentFor(r.assignments(roomID, rm.controller.Snapshot(), nil, "", ""), viewerPeerID).Upstream
}
// peerSignalAuthorization ports peerSignalAuthorization.
// A retired server candidate must never be mistaken for a parent-initiated
// replacement. Its ID remains opaque to clients; no retired-ID history is needed.
const candidateConnectionIDPrefix = "route_"
// peerSignalAuthorization resolves candidate and retained media by connection,
// since both may belong to the same parent/child pair during overlap.
func (r *router) peerSignalAuthorization(input peerSignalInput) signalAuthorization {
rm, _ := r.rooms.Get(input.roomID)
if rm == nil || rm.controller == nil {
@@ -586,7 +591,7 @@ func (r *router) peerSignalAuthorization(input peerSignalInput) signalAuthorizat
input.targetPeerID == operation.ChildPeerID
childToParent := input.sourcePeerID == operation.ChildPeerID &&
input.targetPeerID == current.Tuple.ParentPeerID
if parentToChild || childToParent {
if (parentToChild || childToParent) && input.connectionID == current.ConnectionID {
parentSession, childSession := input.targetSessionID, input.sourceSessionID
if parentToChild {
parentSession, childSession = input.sourceSessionID, input.targetSessionID
@@ -594,7 +599,6 @@ func (r *router) peerSignalAuthorization(input peerSignalInput) signalAuthorizat
if hasParent && hasChild &&
parent.SessionID == parentSession &&
child.SessionID == childSession &&
input.connectionID == current.ConnectionID &&
(input.signalKind == "candidate" ||
(parentToChild && input.descriptionType == "offer") ||
(childToParent && input.descriptionType == "answer")) {
@@ -627,6 +631,9 @@ func (r *router) peerSignalAuthorization(input peerSignalInput) signalAuthorizat
input.signalKind == "description" &&
input.descriptionType == "offer" &&
input.connectionID != edge.ConnectionID {
if strings.HasPrefix(input.connectionID, candidateConnectionIDPrefix) {
return signalAuthorizationDenied
}
adopted := controller.AdoptDirectConnection(route.AdoptDirectConnectionInput{
EdgeGuard: route.EdgeGuard{
ChildPeerID: childPeerID,
@@ -636,8 +643,13 @@ func (r *router) peerSignalAuthorization(input peerSignalInput) signalAuthorizat
ConnectionID: edge.ConnectionID,
},
NewConnectionID: input.connectionID,
})
if adopted {
}, r.now())
if adopted.Accepted {
r.releaseResources(adopted.Released)
if adopted.ActiveRevision != snapshot.Revision {
r.broadcastActive(input.roomID, rm)
r.requestPump(input.roomID)
}
return signalAuthorizationAllowed
}
return signalAuthorizationDenied
internal/server/signal/router_pump.go
+1 -1
View File
@@ -308,7 +308,7 @@ func (r *router) prepareCandidate(
if !hasChild || child.SessionID != operation.ChildSessionID || shareGeneration == "" {
return denied(route.RejectionStale)
}
connectionID := opaqueID()
connectionID := candidateConnectionIDPrefix + opaqueID()
var overlap *route.Resource
if plan.EndpointTransition.Kind == route.TransitionOverlap {
producerPeerID := ""
internal/server/signal/router_test.go
+3 -5
View File
@@ -69,14 +69,13 @@ type routerHarness struct {
func newRouterHarness(t *testing.T, options routerHarnessOptions) *routerHarness {
t.Helper()
h := &routerHarness{t: t, sent: map[string][]protocol.ServerMessage{}}
storeOptions := room.Options{LeaseMs: 60_000, MaxRooms: 4, MaxViewersPerRoom: 20}
storeOptions := room.Options{MaxRooms: 4, MaxViewersPerRoom: 20}
var now func() int64
var afterFunc func(time.Duration, func()) func() bool
if options.fakeTimers {
h.clock = &manualClock{settle: h.settle}
now = h.clock.now
afterFunc = h.clock.afterFunc
storeOptions.Now = now
}
store, err := room.New(storeOptions)
if err != nil {
@@ -156,10 +155,9 @@ func (h *routerHarness) createRoom() room.CreatedRoom {
var created room.CreatedRoom
var err error
h.locked(func() {
var lease int64
lease, err = h.store.BeginCreateRoom()
err = h.store.BeginCreateRoom()
if err == nil {
created, err = h.store.CreateRoom(protocol.CodeEntryOpen, nil, nil, "", lease)
created, err = h.store.CreateRoom(protocol.CodeEntryOpen, nil, nil, "")
}
})
if err != nil {
internal/server/signal/server.go
+10 -37
View File
@@ -58,7 +58,6 @@ const (
defaultAuthenticationTimeoutMs = 5_000
defaultViewerDisconnectGraceMs = 5_000
defaultHeartbeatIntervalMs = 30_000
defaultCleanupIntervalMs = 30_000
)
// SfuFallback is the embedded media runtime the media router
@@ -91,7 +90,6 @@ type Options struct {
AuthenticationTimeoutMs int
ViewerDisconnectGraceMs int
HeartbeatIntervalMs int
CleanupIntervalMs int
MaxConnections int
MaxUnauthenticatedConnections int
// AfterFunc is the timer factory (D5); nil uses time.AfterFunc.
@@ -138,7 +136,7 @@ type viewerQualityEvidenceAttempt struct {
// (authenticate falls back to the session id), which is what made the TS
// `undefined === undefined` comparisons meaningful.
type roomShare struct {
// generation survives stopSharing; only terminateRoom drops the whole
// generation survives stopSharing; only closeRoom drops the whole
// record (hazard 2).
generation string
// qualitySettings and routePolicy are nil where the TS map had no entry;
@@ -182,7 +180,6 @@ type Server struct {
authenticationTimeoutMs int
viewerDisconnectGraceMs int
heartbeatIntervalMs int
cleanupIntervalMs int
maxConnections int
maxUnauthenticatedConnections int
@@ -208,12 +205,11 @@ type Server struct {
deferredViewerPresenceRooms map[string]struct{}
heartbeatStop func() bool
cleanupStop func() bool
closing bool
}
// New is the SignalingServer constructor. It performs no I/O; the heartbeat
// and cleanup tickers start immediately.
// ticker starts immediately.
func New(options Options) (*Server, error) {
if err := protocol.AssertEndpointMediaCopyCapacity(options.EndpointMediaCopyCapacity); err != nil {
return nil, err
@@ -239,7 +235,6 @@ func New(options Options) (*Server, error) {
authenticationTimeoutMs: orDefault(options.AuthenticationTimeoutMs, defaultAuthenticationTimeoutMs),
viewerDisconnectGraceMs: orDefault(options.ViewerDisconnectGraceMs, defaultViewerDisconnectGraceMs),
heartbeatIntervalMs: orDefault(options.HeartbeatIntervalMs, defaultHeartbeatIntervalMs),
cleanupIntervalMs: orDefault(options.CleanupIntervalMs, defaultCleanupIntervalMs),
maxConnections: orDefault(options.MaxConnections, defaultMaxSignalConnections),
maxUnauthenticatedConnections: orDefault(options.MaxUnauthenticatedConnections, defaultMaxUnauthenticatedConnections),
sessionsByID: map[string]*session{},
@@ -295,7 +290,6 @@ func New(options Options) (*Server, error) {
s.router = mediaRouter
s.mu.Lock()
s.armHeartbeat()
s.armCleanup()
s.mu.Unlock()
return s, nil
}
@@ -449,9 +443,6 @@ func (s *Server) Close(ctx context.Context) error {
if s.heartbeatStop != nil {
s.heartbeatStop()
}
if s.cleanupStop != nil {
s.cleanupStop()
}
for key, timer := range s.viewerGraceTimers {
timer.stop()
delete(s.viewerGraceTimers, key)
@@ -639,8 +630,7 @@ func (s *Server) CreateRoom(
) (room.CreatedRoom, error) {
s.mu.Lock()
defer s.mu.Unlock()
lease, err := s.store.BeginCreateRoom()
if err != nil {
if err := s.store.BeginCreateRoom(); err != nil {
return room.CreatedRoom{}, err
}
var material []byte
@@ -652,7 +642,7 @@ func (s *Server) CreateRoom(
material, derived = s.store.DeriveViewerPasswordMaterial(*roomPassword, nil)
}()
}
return s.store.CreateRoom(codeEntryPolicy, material, derived, preferredRoomID, lease)
return s.store.CreateRoom(codeEntryPolicy, material, derived, preferredRoomID)
}
// armHeartbeat is the heartbeat setInterval (T1): each tick re-arms the
@@ -674,19 +664,6 @@ func (s *Server) armHeartbeat() {
})
}
// armCleanup is the expireRooms setInterval (T2).
func (s *Server) armCleanup() {
s.cleanupStop = s.afterFunc(time.Duration(s.cleanupIntervalMs)*time.Millisecond, func() {
s.mu.Lock()
defer s.mu.Unlock()
if s.closing {
return
}
s.expireRooms()
s.armCleanup()
})
}
// ---------------------------------------------------------------------------
// messages
// ---------------------------------------------------------------------------
@@ -911,7 +888,7 @@ func (s *Server) completeAuthentication(sess *session, request authRequest, part
if s.closing || !s.sessions.Has(sess) || sess.revoked || !sess.open() {
if _, err := s.store.DisconnectParticipant(participant.RoomID, participant.PeerID, sess.sessionID); err != nil {
// The TS threw out of the async function; there is no catcher (D7).
panic(fmt.Errorf("room store write failed while releasing an aborted authentication: %w", err))
panic(fmt.Errorf("room store failed while releasing an aborted authentication: %w", err))
}
return
}
@@ -985,7 +962,6 @@ func (s *Server) completeAuthentication(sess *session, request authRequest, part
Type: "authenticated",
Protocol: protocol.SignalingProtocol,
PeerID: participant.PeerID,
RoomExpiresAt: participant.ExpiresAt,
MaxViewers: protocol.Int(s.store.MaxViewersPerRoom()),
EndpointMediaCopyCapacity: protocol.Int(s.endpointMediaCopyCapacity),
HostOnline: participant.HostOnline,
@@ -1009,7 +985,6 @@ func (s *Server) completeAuthentication(sess *session, request authRequest, part
Type: "authenticated",
Protocol: protocol.SignalingProtocol,
PeerID: participant.PeerID,
RoomExpiresAt: participant.ExpiresAt,
MaxViewers: protocol.Int(s.store.MaxViewersPerRoom()),
EndpointMediaCopyCapacity: protocol.Int(s.endpointMediaCopyCapacity),
HostOnline: participant.HostOnline,
@@ -1306,7 +1281,7 @@ func (s *Server) handleAuthenticatedMessage(sess *session, authenticated *authen
if s.sessions.Has(sess) && sess.authenticated == authenticated {
if _, err := s.store.DisconnectParticipant(authenticated.roomID, authenticated.peerID, sess.sessionID); err != nil {
// The TS threw out of the message handler; nothing caught it (D7).
panic(fmt.Errorf("room store write failed while stopping sharing: %w", err))
panic(fmt.Errorf("room store failed while stopping sharing: %w", err))
}
}
s.stopSharing(authenticated.roomID)
@@ -1643,15 +1618,13 @@ func authenticationErrorMessage(code string) string {
case "ROOM_ACCESS_DENIED":
return "Room access denied"
case "ROOM_NOT_FOUND":
return "Room not found or expired"
case "ROOM_EXPIRED":
return "Room has expired"
return "Room not found"
case "ROOM_FULL":
return "Room is full"
case "HOST_ALREADY_CONNECTED":
return "A host is already connected"
case "INVALID_TOKEN":
return "Room is invalid or expired"
return "Room credentials are invalid"
}
return "Authentication failed"
}
@@ -1672,12 +1645,12 @@ func authenticationErrorCode(request authRequest, err error) string {
return string(roomError.Code)
}
if request.viewerGrant == "" {
if roomError.Code == room.CodeRoomNotFound || roomError.Code == room.CodeRoomExpired {
if roomError.Code == room.CodeRoomNotFound {
return "ROOM_NOT_FOUND"
}
return "ROOM_ACCESS_DENIED"
}
if roomError.Code == room.CodeRoomExpired || roomError.Code == room.CodeRoomNotFound {
if roomError.Code == room.CodeRoomNotFound {
return "INVALID_TOKEN"
}
return string(roomError.Code)
internal/server/signal/server_test.go
+56 -115
View File
@@ -48,7 +48,7 @@ func TestCreateRoomReleasesLockWhenDerivationOrCommitPanics(t *testing.T) {
for _, password := range []string{"", "room-password"} {
t.Run(password, func(t *testing.T) {
store, err := room.New(room.Options{
LeaseMs: 86_400_000, MaxRooms: 1, MaxViewersPerRoom: 1,
MaxRooms: 1, MaxViewersPerRoom: 1,
Random: func(int) []byte { panic("test random source") },
})
if err != nil {
@@ -71,46 +71,27 @@ func TestCreateRoomReleasesLockWhenDerivationOrCommitPanics(t *testing.T) {
}
}
func TestIncomingStorageFailureReleasesLockWithoutSuppressingPanic(t *testing.T) {
func TestStopSharingDoesNotWriteRoomAuthority(t *testing.T) {
database, err := room.NewDatabase(filepath.Join(t.TempDir(), "rooms.sqlite"))
if err != nil {
t.Fatal(err)
}
store := newStore(t, 1, nil, database)
defer store.Close()
created := createStoreRoom(t, store, protocol.CodeEntryOpen, "")
sess := newSession(nil, true)
defer sess.cancel()
participant, err := store.ConnectParticipant(room.ConnectParticipantInput{
RoomID: created.RoomID, Role: protocol.RoleHost, Token: created.HostToken,
ClientID: "failed-storage-host", SessionID: sess.sessionID,
store := newStore(t, 1, database)
h := startHarness(t, harnessOptions{store: store})
host := openClient(t, h)
authenticated := authenticate(t, host, h.room, protocol.RoleHost, "storage-free-host", 1, "", presenceOptions{})
h.locked(func() {
if err := database.Close(); err != nil {
t.Fatal(err)
}
})
if err != nil {
t.Fatal(err)
host.sendJSON(map[string]any{"type": "stop-sharing", "shareGeneration": authenticated.ShareGeneration})
expectClose(t, host, 1000, "Sharing stopped")
if h.storeSize() != 1 {
t.Fatal("stopping sharing removed room authority")
}
const generation = "failed_storage_share_generation"
sess.authenticated = &authenticatedSession{roomID: created.RoomID, role: protocol.RoleHost,
peerID: participant.PeerID, shareGeneration: generation}
server := &Server{store: store, shares: map[string]roomShare{created.RoomID: {generation: generation}}}
server.sessions.Set(sess, struct{}{})
if err = database.Close(); err != nil {
t.Fatal(err)
}
func() {
defer func() {
failure, ok := recover().(error)
if !ok || !strings.Contains(failure.Error(), "room store write failed while stopping sharing") {
t.Errorf("handler did not preserve its storage panic: %v", failure)
}
}()
server.handleIncoming(sess, websocket.MessageText,
[]byte(`{"type":"stop-sharing","shareGeneration":"`+generation+`"}`))
}()
if !server.mu.TryLock() {
server.mu.Unlock() // Allow test cleanup if the regression returns.
t.Fatal("storage panic left the message lock held before reader cleanup")
}
server.mu.Unlock()
resumed := openClient(t, h)
authenticate(t, resumed, h.room, protocol.RoleHost, "storage-free-host", 1, "", presenceOptions{})
}
// ---------------------------------------------------------------------------
@@ -128,7 +109,6 @@ type harnessOptions struct {
stunURLs []string
natPredictionEnabled bool
now func() int64
cleanupIntervalMs int
store *room.Store
room *room.CreatedRoom
afterFunc func(time.Duration, func()) func() bool
@@ -152,13 +132,11 @@ func orValue(value, fallback int) int {
return value
}
func newStore(t *testing.T, maxViewersPerRoom int, now func() int64, database *room.Database) *room.Store {
func newStore(t *testing.T, maxViewersPerRoom int, database *room.Database) *room.Store {
t.Helper()
store, err := room.New(room.Options{
LeaseMs: 86_400_000,
MaxRooms: room.Capacity,
MaxViewersPerRoom: maxViewersPerRoom,
Now: now,
Database: database,
})
if err != nil {
@@ -175,8 +153,7 @@ func newStore(t *testing.T, maxViewersPerRoom int, now func() int64, database *r
// createStoreRoom is roomStore.createRoom() on a store no server guards yet.
func createStoreRoom(t *testing.T, store *room.Store, policy protocol.CodeEntryPolicy, password string) room.CreatedRoom {
t.Helper()
lease, err := store.BeginCreateRoom()
if err != nil {
if err := store.BeginCreateRoom(); err != nil {
t.Fatalf("BeginCreateRoom: %v", err)
}
var material []byte
@@ -184,7 +161,7 @@ func createStoreRoom(t *testing.T, store *room.Store, policy protocol.CodeEntryP
if password != "" {
material, derived = store.DeriveViewerPasswordMaterial(password, nil)
}
created, err := store.CreateRoom(policy, material, derived, "", lease)
created, err := store.CreateRoom(policy, material, derived, "")
if err != nil {
t.Fatalf("CreateRoom: %v", err)
}
@@ -196,7 +173,7 @@ func startHarness(t *testing.T, options harnessOptions) *harness {
maxViewersPerRoom := orValue(options.maxViewersPerRoom, 8)
store := options.store
if store == nil {
store = newStore(t, maxViewersPerRoom, options.now, nil)
store = newStore(t, maxViewersPerRoom, nil)
}
var created room.CreatedRoom
if options.room != nil {
@@ -227,7 +204,6 @@ func startHarness(t *testing.T, options harnessOptions) *harness {
AuthenticationTimeoutMs: orValue(options.authenticationTimeoutMs, 500),
ViewerDisconnectGraceMs: orValue(options.viewerDisconnectGraceMs, 50),
HeartbeatIntervalMs: 60_000,
CleanupIntervalMs: orValue(options.cleanupIntervalMs, 60_000),
MaxConnections: options.maxSignalConnections,
MaxUnauthenticatedConnections: options.maxUnauthenticatedSignalConnections,
AfterFunc: options.afterFunc,
@@ -1163,7 +1139,7 @@ func TestSignalRebuildsRouteWhenViewerReconnectsBeforeHostAfterRestart(t *testin
if err != nil {
t.Fatal(err)
}
firstStore := newStore(t, 8, nil, firstDatabase)
firstStore := newStore(t, 8, firstDatabase)
rm := createStoreRoom(t, firstStore, protocol.CodeEntryOpen, "")
first := startHarness(t, harnessOptions{store: firstStore, room: &rm})
firstHost := openClient(t, first)
@@ -1178,7 +1154,7 @@ func TestSignalRebuildsRouteWhenViewerReconnectsBeforeHostAfterRestart(t *testin
if err != nil {
t.Fatal(err)
}
secondStore := newStore(t, 8, nil, secondDatabase)
secondStore := newStore(t, 8, secondDatabase)
second := startHarness(t, harnessOptions{store: secondStore, room: &rm})
secondViewer := openClient(t, second)
viewerAuth := authenticate(t, secondViewer, rm, protocol.RoleViewer, "restart-route-viewer", 1, "", presenceOptions{})
@@ -1913,11 +1889,11 @@ func TestSignalDistinguishesMissingRoomsFromExistingRoomCodeOnlyDenials(t *testi
})
protectedRoom := h.createRoom(protocol.CodeEntryPrivate, "correct-password")
expiring := h.createRoom(protocol.CodeEntryOpen, "")
abandoned := h.createRoom(protocol.CodeEntryOpen, "")
unusedRoomID := ""
for code := 1_000; code < 10_000; code++ {
candidate := fmt.Sprint(code)
if candidate != h.room.RoomID && candidate != protectedRoom.RoomID && candidate != expiring.RoomID {
if candidate != h.room.RoomID && candidate != protectedRoom.RoomID && candidate != abandoned.RoomID {
unusedRoomID = candidate
break
}
@@ -1936,7 +1912,7 @@ func TestSignalDistinguishesMissingRoomsFromExistingRoomCodeOnlyDenials(t *testi
sendCodeOnlyAuthenticate(client, roomID, clientID, extra)
message := "Room access denied"
if code == "ROOM_NOT_FOUND" {
message = "Room not found or expired"
message = "Room not found"
}
expectEqual(t, client.next("error").raw, fmt.Sprintf(`{"type":"error","code":%q,"message":%q}`, code, message))
expectClose(t, client, 4003, "Authentication failed")
@@ -1947,85 +1923,50 @@ func TestSignalDistinguishesMissingRoomsFromExistingRoomCodeOnlyDenials(t *testi
expectDenial(protectedRoom.RoomID, "wrong-password-viewer", "ROOM_ACCESS_DENIED", "wrong-password")
expectDenial(h.room.RoomID, "full-room-viewer", "ROOM_ACCESS_DENIED", "")
now.advance(86_400_001)
expectDenial(expiring.RoomID, "expired-room-viewer", "ROOM_NOT_FOUND", "")
expiredGrant := openClientWithCookie(t, h, siteAccessCookie)
sendCodeOnlyAuthenticate(expiredGrant, expiring.RoomID, "expired-grant-viewer",
map[string]any{"viewerGrant": expiring.ViewerGrant})
expectMatch(t, expiredGrant.next("error").raw, `{"code":"INVALID_TOKEN"}`)
}
// ---------------------------------------------------------------------------
// room lease and expiry
// ---------------------------------------------------------------------------
func TestSignalDoesNotExpireActiveRoomAndArmsItsLeaseAfterHostDisconnect(t *testing.T) {
now := newClock(utcMillis(2026, time.August, 20, 12))
h := startHarness(t, harnessOptions{now: now.now})
host := openClient(t, h)
authenticated := authenticate(t, host, h.room, protocol.RoleHost, "active-lease-host", 1, "", presenceOptions{})
viewer := openClient(t, h)
authenticate(t, viewer, h.room, protocol.RoleViewer, "provisional-viewer", 1, "", presenceOptions{})
commitPreparedRoute(t, viewer)
now.advance(86_401_000)
expireAt := func(nowMs int64) []room.ClosedRoom {
t.Helper()
var expired []room.ClosedRoom
var err error
h.locked(func() { expired, err = h.store.ExpireRooms(nowMs) })
if err != nil {
t.Fatalf("ExpireRooms: %v", err)
h.locked(func() {
if _, err := h.store.AbandonRoom(abandoned.RoomID); err != nil {
t.Fatal(err)
}
return expired
}
if expired := expireAt(now.now()); len(expired) != 0 {
t.Fatalf("active room expired: %v", expired)
}
h.closeClient(host)
viewer.next("host-status")
var hostConnected bool
h.locked(func() { _, hostConnected = h.store.GetConnectedHost(h.room.RoomID) })
expectTrue(t, "host must be disconnected", !hostConnected)
if expired := expireAt(now.now() + 86_399_999); len(expired) != 0 {
t.Fatalf("room expired before its lease: %v", expired)
}
expired := expireAt(now.now() + 86_401_000)
if len(expired) != 1 || expired[0].RoomID != h.room.RoomID || len(expired[0].SessionIDs) != 1 {
t.Fatalf("expired rooms: %+v", expired)
}
h.closeClient(viewer)
expectString(t, "role", authenticated.Role, protocol.RoleHost)
})
expectDenial(abandoned.RoomID, "abandoned-room-viewer", "ROOM_NOT_FOUND", "")
abandonedGrant := openClientWithCookie(t, h, siteAccessCookie)
sendCodeOnlyAuthenticate(abandonedGrant, abandoned.RoomID, "abandoned-grant-viewer",
map[string]any{"viewerGrant": abandoned.ViewerGrant})
expectMatch(t, abandonedGrant.next("error").raw, `{"code":"INVALID_TOKEN"}`)
}
func TestSignalTerminatesExpiredDormantRoomThroughCleanupTimer(t *testing.T) {
// ---------------------------------------------------------------------------
// room lifetime
// ---------------------------------------------------------------------------
func TestSignalRetainsDormantAuthorityAcrossLongHostAbsence(t *testing.T) {
now := newClock(utcMillis(2026, time.August, 20, 12))
h := startHarness(t, harnessOptions{now: now.now, cleanupIntervalMs: 20})
h := startHarness(t, harnessOptions{now: now.now})
now.advance(30 * 24 * 60 * 60 * 1_000)
host := openClient(t, h)
authenticate(t, host, h.room, protocol.RoleHost, "expiring-host", 1, "", presenceOptions{})
authenticate(t, host, h.room, protocol.RoleHost, "returning-host", 1, "", presenceOptions{})
viewer := openClient(t, h)
authenticate(t, viewer, h.room, protocol.RoleViewer, "expiring-viewer", 1, "", presenceOptions{})
commitPreparedRoute(t, viewer)
authenticate(t, viewer, h.room, protocol.RoleViewer, "waiting-viewer", 1, "", presenceOptions{})
h.closeClient(host)
viewer.next("host-status")
now.advance(86_401_000)
expectEqual(t, viewer.next("room-closed").raw, `{"type":"room-closed","reason":"expired"}`)
expectClose(t, viewer, 1000, "Room expired")
if size := h.storeSize(); size != 0 {
t.Fatalf("store size %d", size)
now.advance(30 * 24 * 60 * 60 * 1_000)
resumed := openClient(t, h)
authenticate(t, resumed, h.room, protocol.RoleHost, "returning-host", 1, "", presenceOptions{})
lateViewer := openClient(t, h)
authenticate(t, lateViewer, h.room, protocol.RoleViewer, "late-viewer", 1, "", presenceOptions{})
if h.storeSize() != 1 || viewer.isClosed() {
t.Fatal("host absence retired the room or its Viewer")
}
}
func TestSignalKeepsPreHostViewerSessionCurrentAfterHostClearsRoomLease(t *testing.T) {
func TestSignalKeepsPreHostViewerSessionCurrentAfterHostConnects(t *testing.T) {
now := newClock(utcMillis(2026, time.August, 20, 12))
h := startHarness(t, harnessOptions{now: now.now})
viewer := openClient(t, h)
authenticate(t, viewer, h.room, protocol.RoleViewer, "pre-host-lease-viewer", 1, "", presenceOptions{displayName: "Before"})
authenticate(t, viewer, h.room, protocol.RoleViewer, "pre-host-viewer", 1, "", presenceOptions{displayName: "Before"})
host := openClient(t, h)
authenticate(t, host, h.room, protocol.RoleHost, "pre-host-lease-host", 1, "", presenceOptions{viewerPresence: true})
authenticate(t, host, h.room, protocol.RoleHost, "pre-host-host", 1, "", presenceOptions{viewerPresence: true})
named := func(name string) func(presenceMessage) bool {
return func(message presenceMessage) bool {
return slices.ContainsFunc(viewerPresenceEntries(message), func(entry presenceEntry) bool {
@@ -2055,7 +1996,7 @@ func TestSignalLeavesEstablishedAuthorizationAndMediaUntouchedWhenGrantUpdateFai
if err != nil {
t.Fatal(err)
}
store := newStore(t, 8, nil, database)
store := newStore(t, 8, database)
h := startHarness(t, harnessOptions{store: store})
host := openClient(t, h)
hostAuth := authenticate(t, host, h.room, protocol.RoleHost, "persistence-failure-host", 1, "", presenceOptions{})
@@ -2676,7 +2617,7 @@ func (m *timerQueue) fireWhere(due func(time.Duration) bool) {
func TestSignalHeartbeatTerminatesSilentConnectionAfterMissedPong(t *testing.T) {
timers := &timerQueue{}
h := startHarness(t, harnessOptions{afterFunc: timers.afterFunc, cleanupIntervalMs: 30_000})
h := startHarness(t, harnessOptions{afterFunc: timers.afterFunc})
heartbeat := 60_000 * time.Millisecond
responsive := openClient(t, h)
header := http.Header{}