fix: 11 defects from bughunt sweep across server, db and client (#1382)

* fix(ws): 1 defect(s) (OC-0001)

* fix(db): 1 defect(s) (OC-0002)

sanitizeFTSQuery filtered only characters, so FTS5's bareword boolean
keywords (AND, OR, NOT) reached MATCH as operators; a query in an
invalid operator position raised "fts5: syntax error" instead of
returning results. Drop those bareword tokens after sanitizing.

* fix(dm): 1 defect(s) (OC-0004)

* fix(ws): 1 defect(s) (OC-0005)

* fix(voice): 1 defect(s) (OC-0006)

Count the shared voice_max_video budget in streams rather than rows: a
single user publishing both camera and screenshare consumed one slot while
producing two live streams, letting a channel over-admit up to 2N streams
against an N-stream cap.

* fix(client): 2 defect(s) (OC-0007, OC-0009)

OC-0007: mark the active channel loading before invalidating its message
window on a full-ready resync, so MessageList shows the spinner instead
of the empty-channel state for the duration of the refetch.

OC-0009: fan USER_UPDATE renames out to voiceStore.voiceUsers, which
keeps its own frozen username copy, so the voice roster no longer shows
a stale name for the rest of the call.

* fix(admin): 1 defect(s) (OC-0010)

* fix(identity): 1 defect(s) (OC-0011)

* fix(ws): 1 defect(s) (OC-0003)

The public half of an invisible user's presence (PresenceOthersEvent, and
BroadcastPresence's own mapped payload) went out via broadcastExcludeLow on
the low-priority queue - the ephemeral, unsequenced, drop-on-overflow
transport built for typing indicators - while every other source of the same
user's presence shares the normal-priority queue. That split one user's
presence across two per-client FIFOs with different durability and different
drain order (writePump drains normal strictly before low), so a frame could
land out of order against a later connect/disconnect presence frame, or be
silently dropped with no replay recovery.

Adds Hub.BroadcastToAllExcept, which routes through the same h.broadcast
channel and seqMu-serialized deliverBroadcast as BroadcastToAll, carrying an
excludeUserID that deliverBroadcast applies via pubsub.Publish(TopicGlobal,
msg, excludeUserID).

* fix(ws): 1 defect(s) (OC-0008)

---------

Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
J3vb
2026-08-16 17:06:11 +02:00
committed by GitHub
co-authored by Claude
parent d6c768cb90
commit 36be31db43
25 changed files with 1346 additions and 62 deletions
+14
View File
@@ -29,6 +29,7 @@ import {
messagesStore, messagesStore,
setMessages, setMessages,
invalidateLoadedMessageWindows, invalidateLoadedMessageWindows,
setChannelLoading,
setChannelLoadError, setChannelLoadError,
} from "@stores/messages.store"; } from "@stores/messages.store";
import { import {
@@ -44,6 +45,7 @@ import {
voiceStore, voiceStore,
setVoiceStates, setVoiceStates,
updateVoiceState, updateVoiceState,
updateVoiceUserProfile,
removeVoiceUser, removeVoiceUser,
setVoiceConfig, setVoiceConfig,
setSpeakers, setSpeakers,
@@ -364,6 +366,13 @@ export function wireDispatcher(
// showing only carried-through pending rows until the user navigates // showing only carried-through pending rows until the user navigates
// away and back. // away and back.
if (activeAfterReady !== null && getMessages !== undefined) { if (activeAfterReady !== null && getMessages !== undefined) {
// Mark the active channel loading BEFORE invalidating its window —
// invalidate drops its rows synchronously, and if historyLoadState
// is left idle for even one microtask, MessageList's "no rows +
// idle" empty-state branch renders the channel as genuinely empty
// for the whole in-flight refetch instead of showing the in-region
// spinner (OC-0007).
setChannelLoading(activeAfterReady);
invalidateLoadedMessageWindows(); invalidateLoadedMessageWindows();
getMessages(activeAfterReady, { limit: 50 }) getMessages(activeAfterReady, { limit: 50 })
.then((resp) => { .then((resp) => {
@@ -797,6 +806,11 @@ export function wireDispatcher(
avatar: payload.avatar ?? "", avatar: payload.avatar ?? "",
...(payload.display_name === undefined ? {} : { displayName: payload.display_name ?? "" }), ...(payload.display_name === undefined ? {} : { displayName: payload.display_name ?? "" }),
}); });
// voiceStore.voiceUsers is the third store holding a frozen username
// copy (see updateVoiceUserProfile's doc comment) — without this, a
// rename leaves the voice roster showing the old name for the rest of
// the call.
updateVoiceUserProfile(payload.user_id, { username: payload.username });
// Update auth store if the current user changed their own profile. // Update auth store if the current user changed their own profile.
const currentUser = authStore.getState().user; const currentUser = authStore.getState().user;
@@ -244,6 +244,28 @@ export function updateVoiceState(payload: VoiceStatePayload): void {
}); });
} }
/** Patch a user's username in every voice channel they currently occupy.
* VoiceUser keeps its own frozen copy of the username (the same shape as
* membersStore's and dmStore's copies), refreshed only by setVoiceStates
* (ready) and updateVoiceState (voice_state) — a profile rename via
* USER_UPDATE must patch it too, or the voice roster keeps showing the old
* name for the rest of the call, until the user leaves and rejoins voice. */
export function updateVoiceUserProfile(userId: number, patch: { readonly username: string }): void {
voiceStore.setState((prev) => {
let changed = false;
const nextChannels = new Map(prev.voiceUsers);
for (const [channelId, users] of prev.voiceUsers) {
const existing = users.get(userId);
if (existing === undefined) continue;
const nextUsers = new Map(users);
nextUsers.set(userId, { ...existing, username: patch.username });
nextChannels.set(channelId, nextUsers);
changed = true;
}
return changed ? { ...prev, voiceUsers: nextChannels } : prev;
});
}
/** Remove a user from a voice channel. */ /** Remove a user from a voice channel. */
export function removeVoiceUser(payload: VoiceLeavePayload): void { export function removeVoiceUser(payload: VoiceLeavePayload): void {
voiceStore.setState((prev) => { voiceStore.setState((prev) => {
@@ -695,6 +695,44 @@ describe("WS Dispatcher", () => {
expect(member?.avatar).toBe("/api/v1/files/abc"); expect(member?.avatar).toBe("/api/v1/files/abc");
}); });
// OC-0009: voiceStore.voiceUsers keeps its own frozen copy of each
// participant's username, written only by setVoiceStates (ready) and
// updateVoiceState (voice_state). USER_UPDATE fans a profile change out to
// membersStore and dmStore, which hold the same kind of frozen copy, but
// not to voiceStore — so a rename leaves the voice roster showing the old
// username for the rest of the call.
it("updates the voice roster username on user_update", () => {
voiceStore.setState((prev) => {
const userMap = new Map([
[
10,
{
userId: 10,
username: "bob",
muted: false,
deafened: false,
speaking: false,
camera: false,
screenshare: false,
},
],
]);
const voiceUsers = new Map(prev.voiceUsers);
voiceUsers.set(5, userMap);
return { ...prev, voiceUsers };
});
mock.dispatch("user_update", {
user_id: 10,
username: "bobby",
avatar: "new.png",
display_name: null,
});
const voiceUser = voiceStore.getState().voiceUsers.get(5)?.get(10);
expect(voiceUser?.username).toBe("bobby");
});
describe("presence and user_update sync dmStore", () => { describe("presence and user_update sync dmStore", () => {
const dmChannel = { const dmChannel = {
channelId: 50, channelId: 50,
@@ -1378,6 +1416,54 @@ describe("WS Dispatcher", () => {
expect(isChannelLoaded(1)).toBe(true); expect(isChannelLoaded(1)).toBe(true);
}); });
// OC-0007: invalidateLoadedMessageWindows() synchronously drops the
// active channel's rows and loadedChannels entry, but leaves
// historyLoadState untouched. If the resync handler doesn't also mark
// the channel "loading", MessageList's empty-state branch (no rows +
// idle load state) reads as a genuinely empty channel for the entire
// round trip of the refetch, instead of showing the in-region spinner
// every other first-page fetch shows.
it("marks the active channel loading (not idle) while the resync refetch is in flight", () => {
cleanup();
const listBlocks = vi.fn().mockResolvedValue({ blocked_user_ids: [] });
// Never resolves within this test — pins the synchronous state right
// after the resync `ready` is handled, before the refetch lands.
const getMessages = vi.fn().mockReturnValue(new Promise(() => {}));
cleanup = wireDispatcher(mock.ws, { listBlocks, getMessages });
channelsStore.setState((prev) => ({ ...prev, activeChannelId: 1 }));
setMessages(1, [storedMessage(10)], false);
const readyChannels = [
{ id: 1, name: "general", type: "text" as const, category: null, position: 0 },
];
// First ready: initial connect.
mock.dispatch("ready", {
channels: readyChannels,
members: [],
voice_states: [],
roles: [],
dm_channels: [],
});
// Second ready: a full-ready resync. The synchronous invalidation runs
// and the refetch is fired but never resolves in this test.
mock.dispatch("ready", {
channels: readyChannels,
members: [],
voice_states: [],
roles: [],
dm_channels: [],
});
// The window was dropped...
expect(isChannelLoaded(1)).toBe(false);
expect(getChannelMessages(1)).toEqual([]);
// ...but the load state must be "loading", not idle (null) — idle+empty
// is exactly the shape MessageList reads as "channel has no history".
expect(getHistoryLoadState(1)).toBe("loading");
});
// OC-0203: the refetch above is fired-and-forgotten against whatever // OC-0203: the refetch above is fired-and-forgotten against whatever
// channel was active when the resync `ready` arrived — but the user can // channel was active when the resync `ready` arrived — but the user can
// switch channels before the HTTP response lands. The continuation must // switch channels before the HTTP response lands. The continuation must
+54
View File
@@ -1436,6 +1436,60 @@ func TestAdminAPI_DeleteChannel_NilHubDoesNotPanic(t *testing.T) {
} }
} }
// OC-0010: handleDeleteChannel commits archived=1 as its own transaction and
// evicts voice participants BEFORE calling AdminDeleteChannel — all on
// r.Context(). If the admin's browser aborts the request in that window (tab
// close, navigation, network blip), r.Context() is canceled and the final
// AdminDeleteChannel call fails with context.Canceled: the handler 500s, but
// the archive and the voice eviction already committed. Nothing reverts the
// archive, nothing broadcasts it, and no audit row is written — the channel
// is left silently archived (writes refused, channel_focus 403s) while every
// connected client still shows it live in the sidebar.
//
// This reproduces the race deterministically by canceling the request
// context from inside CleanupVoiceForChannel — exactly the call the repro
// says the real-world cancellation lands during — instead of relying on
// timing. The fix must make the delete tolerate a caller cancellation that
// arrives after the archive has already committed.
func TestAdminAPI_DeleteChannel_SurvivesContextCancelAfterArchiveCommits(t *testing.T) {
database := openAdminTestDB(t)
hub := &mockHub{}
handler := admin.NewAdminAPI(database, "1.0.0", hub, nil, nil, nil, nil, newTestModService(database), newTestRoleService(database))
token := createAdminUser(t, database)
chID, _ := database.AdminCreateChannel(context.Background(), "del-cancel-race", "text", "", "", 0)
ctx, cancel := context.WithCancel(context.Background())
// Fires synchronously inside handleDeleteChannel, after the archive
// commit but before the final AdminDeleteChannel call — the same window
// the repro describes a browser abort landing in.
hub.onVoiceCleanup = func(int64) {
cancel()
}
req := httptest.NewRequest(http.MethodDelete, "/channels/"+itoa(chID), nil)
req = req.WithContext(ctx)
req.Header.Set("Authorization", "Bearer "+token)
w := httptest.NewRecorder()
handler.ServeHTTP(w, req)
if w.Code != http.StatusNoContent {
t.Fatalf("status = %d, want 204 (delete must survive a caller cancellation that arrives after the archive already committed); body: %s", w.Code, w.Body.String())
}
ch, err := database.GetChannel(context.Background(), chID)
if err != nil {
t.Fatalf("GetChannel: %v", err)
}
if ch != nil {
t.Errorf("channel %d still exists after a reported-successful delete: %+v", chID, ch)
}
if len(hub.channelDeleteIDs) != 1 || hub.channelDeleteIDs[0] != chID {
t.Errorf("BroadcastChannelDelete calls = %v, want exactly [%d]", hub.channelDeleteIDs, chID)
}
}
// ─── API tokens: /admin/api/tokens ─────────────────────────────────────────── // ─── API tokens: /admin/api/tokens ───────────────────────────────────────────
func TestAdminAPI_CreateAPIToken_OK(t *testing.T) { func TestAdminAPI_CreateAPIToken_OK(t *testing.T) {
+22 -2
View File
@@ -317,13 +317,33 @@ func handleDeleteChannel(database *db.DB, hub HubBroadcaster) http.HandlerFunc {
hub.CleanupVoiceForChannel(id) hub.CleanupVoiceForChannel(id)
} }
if err := database.AdminDeleteChannel(r.Context(), id); err != nil { // From here on the archive (and, if it ran, the voice eviction)
// already committed. If the admin's browser goes away in this window
// (tab close, navigation, network blip), r.Context() cancels, and an
// AdminDeleteChannel that still used it would fail with
// context.Canceled — 500ing while leaving the channel silently
// archived, unbroadcast and unaudited (OC-0010). Run the rest of the
// delete on an uncancellable tail, matching the repo's convention
// for other durable side effects (totp_handler.go, service/user.go).
delCtx := context.WithoutCancel(r.Context())
if err := database.AdminDeleteChannel(delCtx, id); err != nil {
// A genuine failure here (not caller cancellation, which delCtx
// already absorbs) still leaves the archive committed — tell
// connected clients about the state that did change instead of
// leaving them stuck seeing a live channel that now 403s on
// every read and write.
if hub != nil && !existing.Archived {
if archived, gErr := database.GetChannel(delCtx, id); gErr == nil && archived != nil {
hub.BroadcastChannelUpdate(archived)
hub.RefreshChannelVisibility(archived)
}
}
writeErr(w, http.StatusInternalServerError, "INTERNAL_ERROR", "failed to delete channel") writeErr(w, http.StatusInternalServerError, "INTERNAL_ERROR", "failed to delete channel")
return return
} }
actor := actorFromContext(r) actor := actorFromContext(r)
slog.Warn("channel deleted", "actor_id", actor, "channel_id", id, "name", existing.Name) slog.Warn("channel deleted", "actor_id", actor, "channel_id", id, "name", existing.Name)
db.WriteAudit(context.WithoutCancel(r.Context()), database, actor, "channel_delete", "channel", id, db.WriteAudit(delCtx, database, actor, "channel_delete", "channel", id,
fmt.Sprintf("deleted #%s", existing.Name)) fmt.Sprintf("deleted #%s", existing.Name))
if hub != nil { if hub != nil {
hub.BroadcastChannelDelete(id) hub.BroadcastChannelDelete(id)
+56 -10
View File
@@ -225,6 +225,30 @@ func handleEnableTOTP(pendingStore *auth.PendingTOTPStore, limiter *auth.RateLim
} }
} }
// revokeOtherSessionsAfterAuthChange revokes every session for userID except
// keepSessionID as the security tail of a committed 2FA state change. It
// mirrors UserService.ChangePassword (service/user.go:262-274): a failure is
// logged and retried once (bounded compensating retry for transient write
// contention); if the retry also fails, revoked reports what did succeed and
// failed is true so the caller can report a partial success instead of
// silently claiming the other sessions were revoked when they were not.
func revokeOtherSessionsAfterAuthChange(ctx context.Context, database *db.DB, userID, keepSessionID int64, action string) (revoked int64, failed bool) {
revoked, err := database.DeleteOtherSessions(ctx, userID, keepSessionID)
if err != nil {
slog.Error("DeleteOtherSessions after "+action, "err", err, "user_id", userID)
revokedRetry, retryErr := database.DeleteOtherSessions(ctx, userID, keepSessionID)
if retryErr != nil {
slog.Error("DeleteOtherSessions retry after "+action, "err", retryErr, "user_id", userID)
return revoked, true
}
revoked += revokedRetry
}
if revoked > 0 {
slog.Info("revoked other sessions after "+action, "user_id", userID, "revoked", revoked)
}
return revoked, false
}
func handleConfirmTOTP(database *db.DB, pendingStore *auth.PendingTOTPStore, usedTOTPCodes *auth.UsedTOTPCodeStore, limiter *auth.RateLimiter, totpKey []byte) http.HandlerFunc { func handleConfirmTOTP(database *db.DB, pendingStore *auth.PendingTOTPStore, usedTOTPCodes *auth.UsedTOTPCodeStore, limiter *auth.RateLimiter, totpKey []byte) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) { return func(w http.ResponseWriter, r *http.Request) {
user, ok := r.Context().Value(UserKey).(*db.User) user, ok := r.Context().Value(UserKey).(*db.User)
@@ -313,15 +337,26 @@ func handleConfirmTOTP(database *db.DB, pendingStore *auth.PendingTOTPStore, use
} }
// Security tail of the 2FA change: once the secret update committed, // Security tail of the 2FA change: once the secret update committed,
// revoking the other sessions must not be aborted by a dead request. // revoking the other sessions must not be aborted by a dead request.
n, _ := database.DeleteOtherSessions(context.WithoutCancel(r.Context()), user.ID, keepSessionID) tailCtx := context.WithoutCancel(r.Context())
if n > 0 { revoked, revokeFailed := revokeOtherSessionsAfterAuthChange(tailCtx, database, user.ID, keepSessionID, "totp enable")
slog.Info("revoked other sessions after totp enable", "user_id", user.ID, "revoked", n)
}
slog.Info("totp enabled", "user_id", user.ID) slog.Info("totp enabled", "user_id", user.ID)
db.WriteAudit(context.WithoutCancel(r.Context()), database, user.ID, "totp_enabled", "user", user.ID, db.WriteAudit(tailCtx, database, user.ID, "totp_enabled", "user", user.ID,
"two-factor authentication enrolled") "two-factor authentication enrolled")
if revokeFailed {
// Partial success: 2FA IS enabled; only revoking the other
// sessions failed. A 5xx here would be a lie — the state change
// already committed — so mirror the ChangePassword contract
// (api/profile_handler.go) and report 200 with an explicit warning
// instead of a silent, unqualified 204.
writeJSON(w, http.StatusOK, map[string]any{
"warning": "two-factor authentication enabled, but other sessions could not be revoked; revoke them from the sessions list",
"sessions_revoked": revoked,
})
return
}
w.WriteHeader(http.StatusNoContent) w.WriteHeader(http.StatusNoContent)
} }
} }
@@ -403,15 +438,26 @@ func handleDisableTOTP(database *db.DB, pendingStore *auth.PendingTOTPStore, lim
} }
// Security tail of the 2FA change: once the secret update committed, // Security tail of the 2FA change: once the secret update committed,
// revoking the other sessions must not be aborted by a dead request. // revoking the other sessions must not be aborted by a dead request.
n, _ := database.DeleteOtherSessions(context.WithoutCancel(r.Context()), user.ID, keepSessionID) tailCtx := context.WithoutCancel(r.Context())
if n > 0 { revoked, revokeFailed := revokeOtherSessionsAfterAuthChange(tailCtx, database, user.ID, keepSessionID, "totp disable")
slog.Info("revoked other sessions after totp disable", "user_id", user.ID, "revoked", n)
}
slog.Info("totp disabled", "user_id", user.ID) slog.Info("totp disabled", "user_id", user.ID)
db.WriteAudit(context.WithoutCancel(r.Context()), database, user.ID, "totp_disabled", "user", user.ID, db.WriteAudit(tailCtx, database, user.ID, "totp_disabled", "user", user.ID,
"two-factor authentication disabled") "two-factor authentication disabled")
if revokeFailed {
// Partial success: 2FA IS disabled; only revoking the other
// sessions failed. A 5xx here would be a lie — the state change
// already committed — so mirror the ChangePassword contract
// (api/profile_handler.go) and report 200 with an explicit warning
// instead of a silent, unqualified 204.
writeJSON(w, http.StatusOK, map[string]any{
"warning": "two-factor authentication disabled, but other sessions could not be revoked; revoke them from the sessions list",
"sessions_revoked": revoked,
})
return
}
w.WriteHeader(http.StatusNoContent) w.WriteHeader(http.StatusNoContent)
} }
} }
+144
View File
@@ -528,6 +528,150 @@ func TestDisableTOTP_APITokenPrincipal_RevokesAllSessions(t *testing.T) {
} }
} }
// ─── OC-0011: DeleteOtherSessions failures must not be silent successes ──────
// TestConfirmTOTP_RevokeFailureSurfacesWarning locks down that when
// DeleteOtherSessions fails after a 2FA enable commits, the handler must not
// report unqualified success: it should mirror the ChangePassword contract
// (service/user.go:262-274 / api/profile_handler.go:377) and report a 200
// with an explicit warning, not a silent 204.
func TestConfirmTOTP_RevokeFailureSurfacesWarning(t *testing.T) {
database := newAuthTestDB(t)
limiter := auth.NewRateLimiter()
router := buildAuthRouter(database, limiter)
token := loginAndGetToken(t, router, database, "confirmrevokefail", 4)
user, _ := database.GetUserByUsername(context.Background(), "confirmrevokefail")
if user == nil {
t.Fatal("user not found")
}
// A second, unrelated session so DeleteOtherSessions has a row to delete
// (a DELETE that matches zero rows never fires a BEFORE DELETE trigger).
otherToken, _ := auth.GenerateToken()
if _, err := database.CreateSession(context.Background(), user.ID, auth.HashToken(otherToken), "other-device", "127.0.0.1"); err != nil {
t.Fatalf("CreateSession: %v", err)
}
// Step 1: enable to get a pending secret (before the trigger exists).
rr := postJSONWithToken(t, router, "/api/v1/users/me/totp/enable", token,
map[string]string{"password": "Password1!"})
if rr.Code != http.StatusOK {
t.Fatalf("enable: status = %d; body = %s", rr.Code, rr.Body.String())
}
var enableResp map[string]any
_ = json.NewDecoder(rr.Body).Decode(&enableResp)
secret := extractSecretFromURI(t, enableResp["qr_uri"].(string))
code, _ := auth.GenerateTOTPCode(secret, time.Now().UTC())
// Make every DELETE against sessions fail from here on, simulating a
// genuine DB-level failure (disk error, "database is closed" during
// shutdown) rather than a cancel race — the confirm call below already
// uses context.WithoutCancel, so a dead request cannot trigger this.
if _, err := database.ExecContext(context.Background(), `
CREATE TRIGGER block_delete_sessions
BEFORE DELETE ON sessions
BEGIN
SELECT RAISE(FAIL, 'delete blocked');
END;
`); err != nil {
t.Fatalf("create trigger: %v", err)
}
// Step 2: confirm. The secret update must still commit even though the
// session-revocation tail fails.
rr = postJSONWithToken(t, router, "/api/v1/users/me/totp/confirm", token,
map[string]string{"password": "Password1!", "code": code})
updated, _ := database.GetUserByUsername(context.Background(), "confirmrevokefail")
if updated == nil || updated.TOTPSecret == nil {
t.Fatal("expected TOTPSecret to be committed even though revocation failed")
}
if rr.Code != http.StatusOK {
t.Fatalf("confirm-totp with revoke failure: status = %d, want 200; body = %s", rr.Code, rr.Body.String())
}
var resp map[string]any
if err := json.NewDecoder(rr.Body).Decode(&resp); err != nil {
t.Fatalf("decode confirm response: %v", err)
}
if resp["warning"] == nil {
t.Error("expected a warning field when session revocation failed after totp enable")
}
// The other session must still be present — revocation genuinely failed,
// it was not silently skipped or falsely reported as revoked.
if s, _ := database.GetSessionByTokenHash(context.Background(), auth.HashToken(otherToken)); s == nil {
t.Error("other session should still exist: DeleteOtherSessions was blocked by the trigger")
}
}
// TestDisableTOTP_RevokeFailureSurfacesWarning is the handleDisableTOTP
// sibling of TestConfirmTOTP_RevokeFailureSurfacesWarning.
func TestDisableTOTP_RevokeFailureSurfacesWarning(t *testing.T) {
database := newAuthTestDB(t)
limiter := auth.NewRateLimiter()
router := buildAuthRouter(database, limiter)
token := loginAndGetToken(t, router, database, "disablerevokefail", 4)
user, _ := database.GetUserByUsername(context.Background(), "disablerevokefail")
if user == nil {
t.Fatal("user not found")
}
// Enable and confirm TOTP first (before the trigger exists).
rr := postJSONWithToken(t, router, "/api/v1/users/me/totp/enable", token,
map[string]string{"password": "Password1!"})
var enableResp map[string]any
_ = json.NewDecoder(rr.Body).Decode(&enableResp)
secret := extractSecretFromURI(t, enableResp["qr_uri"].(string))
code, _ := auth.GenerateTOTPCode(secret, time.Now().UTC())
rr = postJSONWithToken(t, router, "/api/v1/users/me/totp/confirm", token,
map[string]string{"password": "Password1!", "code": code})
if rr.Code != http.StatusNoContent {
t.Fatalf("setup confirm: status = %d; body = %s", rr.Code, rr.Body.String())
}
// A second, unrelated session so DeleteOtherSessions has a row to delete.
otherToken, _ := auth.GenerateToken()
if _, err := database.CreateSession(context.Background(), user.ID, auth.HashToken(otherToken), "other-device", "127.0.0.1"); err != nil {
t.Fatalf("CreateSession: %v", err)
}
if _, err := database.ExecContext(context.Background(), `
CREATE TRIGGER block_delete_sessions
BEFORE DELETE ON sessions
BEGIN
SELECT RAISE(FAIL, 'delete blocked');
END;
`); err != nil {
t.Fatalf("create trigger: %v", err)
}
rr = deleteWithToken(t, router, "/api/v1/users/me/totp", token,
map[string]string{"password": "Password1!"})
updated, _ := database.GetUserByUsername(context.Background(), "disablerevokefail")
if updated == nil || updated.TOTPSecret != nil {
t.Fatal("expected TOTPSecret to be cleared even though revocation failed")
}
if rr.Code != http.StatusOK {
t.Fatalf("disable-totp with revoke failure: status = %d, want 200; body = %s", rr.Code, rr.Body.String())
}
var resp map[string]any
if err := json.NewDecoder(rr.Body).Decode(&resp); err != nil {
t.Fatalf("decode disable response: %v", err)
}
if resp["warning"] == nil {
t.Error("expected a warning field when session revocation failed after totp disable")
}
if s, _ := database.GetSessionByTokenHash(context.Background(), auth.HashToken(otherToken)); s == nil {
t.Error("other session should still exist: DeleteOtherSessions was blocked by the trigger")
}
}
// ─── Helpers ───────────────────────────────────────────────────────────────── // ─── Helpers ─────────────────────────────────────────────────────────────────
// deleteWithToken sends a DELETE request with a JSON body and auth token. // deleteWithToken sends a DELETE request with a JSON body and auth token.
+8 -4
View File
@@ -70,10 +70,14 @@ type Querier interface {
DeleteSessionByToken(ctx context.Context, token string) error DeleteSessionByToken(ctx context.Context, token string) error
DisablePlugin(ctx context.Context, id int64) error DisablePlugin(ctx context.Context, id int64) error
EditMessageContent(ctx context.Context, arg EditMessageContentParams) (Message, error) EditMessageContent(ctx context.Context, arg EditMessageContentParams) (Message, error)
// Camera and screenshare share one voice_max_video budget: a channel capped // Camera and screenshare share one voice_max_video budget, counted in
// at N simultaneous video streams must not let a camera publish ignore // STREAMS, not rows: a channel capped at N simultaneous video streams must
// screenshare occupants (or vice versa), so both gates count the same // not let a camera publish ignore screenshare occupants (or vice versa,
// `camera = 1 OR screenshare = 1` slot usage (OC-0023). // OC-0023), and a single user with both flags set must consume two of the N
// slots, not one (OC-0006) -- so both gates sum `vs2.camera + vs2.screenshare`
// across the channel's rows rather than counting rows where either is set.
// The enabling user's own bit is still 0 at gate time, so no self-exclusion
// term is needed.
EnableCameraIfUnderLimit(ctx context.Context, arg EnableCameraIfUnderLimitParams) (sql.Result, error) EnableCameraIfUnderLimit(ctx context.Context, arg EnableCameraIfUnderLimitParams) (sql.Result, error)
EnablePlugin(ctx context.Context, id int64) error EnablePlugin(ctx context.Context, id int64) error
EnableScreenshareIfUnderLimit(ctx context.Context, arg EnableScreenshareIfUnderLimitParams) (sql.Result, error) EnableScreenshareIfUnderLimit(ctx context.Context, arg EnableScreenshareIfUnderLimitParams) (sql.Result, error)
+14 -10
View File
@@ -102,40 +102,44 @@ const enableCameraIfUnderLimit = `-- name: EnableCameraIfUnderLimit :execresult
UPDATE voice_states SET camera = 1 UPDATE voice_states SET camera = 1
WHERE voice_states.user_id = ? AND voice_states.channel_id = ? WHERE voice_states.user_id = ? AND voice_states.channel_id = ?
AND (SELECT COUNT(*) FROM voice_states AS vs2 WHERE vs2.channel_id = ? AND (vs2.camera = 1 OR vs2.screenshare = 1)) < ? AND (SELECT COALESCE(SUM(vs2.camera), 0) + COALESCE(SUM(vs2.screenshare), 0) FROM voice_states AS vs2 WHERE vs2.channel_id = ?) < ?4
` `
type EnableCameraIfUnderLimitParams struct { type EnableCameraIfUnderLimitParams struct {
UserID int64 `json:"userId"` UserID int64 `json:"userId"`
ChannelID int64 `json:"channelId"` ChannelID int64 `json:"channelId"`
ChannelID_2 int64 `json:"channelId2"` ChannelID_2 int64 `json:"channelId2"`
ChannelID_3 int64 `json:"channelId3"` MaxVideo int64 `json:"maxVideo"`
} }
// Camera and screenshare share one voice_max_video budget: a channel capped // Camera and screenshare share one voice_max_video budget, counted in
// at N simultaneous video streams must not let a camera publish ignore // STREAMS, not rows: a channel capped at N simultaneous video streams must
// screenshare occupants (or vice versa), so both gates count the same // not let a camera publish ignore screenshare occupants (or vice versa,
// `camera = 1 OR screenshare = 1` slot usage (OC-0023). // OC-0023), and a single user with both flags set must consume two of the N
// slots, not one (OC-0006) -- so both gates sum `vs2.camera + vs2.screenshare`
// across the channel's rows rather than counting rows where either is set.
// The enabling user's own bit is still 0 at gate time, so no self-exclusion
// term is needed.
func (q *Queries) EnableCameraIfUnderLimit(ctx context.Context, arg EnableCameraIfUnderLimitParams) (sql.Result, error) { func (q *Queries) EnableCameraIfUnderLimit(ctx context.Context, arg EnableCameraIfUnderLimitParams) (sql.Result, error) {
return q.db.ExecContext(ctx, enableCameraIfUnderLimit, return q.db.ExecContext(ctx, enableCameraIfUnderLimit,
arg.UserID, arg.UserID,
arg.ChannelID, arg.ChannelID,
arg.ChannelID_2, arg.ChannelID_2,
arg.ChannelID_3, arg.MaxVideo,
) )
} }
const enableScreenshareIfUnderLimit = `-- name: EnableScreenshareIfUnderLimit :execresult const enableScreenshareIfUnderLimit = `-- name: EnableScreenshareIfUnderLimit :execresult
UPDATE voice_states SET screenshare = 1 UPDATE voice_states SET screenshare = 1
WHERE voice_states.user_id = ? AND voice_states.channel_id = ? WHERE voice_states.user_id = ? AND voice_states.channel_id = ?
AND (SELECT COUNT(*) FROM voice_states AS vs2 WHERE vs2.channel_id = ? AND (vs2.camera = 1 OR vs2.screenshare = 1)) < ? AND (SELECT COALESCE(SUM(vs2.camera), 0) + COALESCE(SUM(vs2.screenshare), 0) FROM voice_states AS vs2 WHERE vs2.channel_id = ?) < ?4
` `
type EnableScreenshareIfUnderLimitParams struct { type EnableScreenshareIfUnderLimitParams struct {
UserID int64 `json:"userId"` UserID int64 `json:"userId"`
ChannelID int64 `json:"channelId"` ChannelID int64 `json:"channelId"`
ChannelID_2 int64 `json:"channelId2"` ChannelID_2 int64 `json:"channelId2"`
ChannelID_3 int64 `json:"channelId3"` MaxVideo int64 `json:"maxVideo"`
} }
func (q *Queries) EnableScreenshareIfUnderLimit(ctx context.Context, arg EnableScreenshareIfUnderLimitParams) (sql.Result, error) { func (q *Queries) EnableScreenshareIfUnderLimit(ctx context.Context, arg EnableScreenshareIfUnderLimitParams) (sql.Result, error) {
@@ -143,7 +147,7 @@ func (q *Queries) EnableScreenshareIfUnderLimit(ctx context.Context, arg EnableS
arg.UserID, arg.UserID,
arg.ChannelID, arg.ChannelID,
arg.ChannelID_2, arg.ChannelID_2,
arg.ChannelID_3, arg.MaxVideo,
) )
} }
+19 -1
View File
@@ -34,6 +34,15 @@ func messageFromGen(m dbgen.Message) *Message {
// ("-col: expr"), so keeping it turns "well-known" into a filter on a // ("-col: expr"), so keeping it turns "well-known" into a filter on a
// nonexistent column "known" and SQLite errors instead of matching. Folding // nonexistent column "known" and SQLite errors instead of matching. Folding
// to a space (rather than dropping it) still matches the indexed tokens. // to a space (rather than dropping it) still matches the indexed tokens.
//
// Filtering characters alone is not enough: FTS5's MATCH grammar also
// recognizes bareword keywords -- AND, OR, NOT (uppercase only) -- as
// boolean operators rather than search terms. Those are ordinary letters, so
// the character filter lets them through unchanged, and a query that places
// one in an invalid position (e.g. the whole query is "AND", or it starts or
// ends with one) makes SQLite raise "fts5: syntax error" instead of running
// the search. Any such token is dropped below so the result is always a
// plain sequence of bareword terms.
func sanitizeFTSQuery(q string) string { func sanitizeFTSQuery(q string) string {
var sb strings.Builder var sb strings.Builder
sb.Grow(len(q)) sb.Grow(len(q))
@@ -51,7 +60,16 @@ func sanitizeFTSQuery(q string) string {
if runes := []rune(result); len(runes) > 200 { if runes := []rune(result); len(runes) > 200 {
result = string(runes[:200]) result = string(runes[:200])
} }
return result
fields := strings.Fields(result)
kept := fields[:0]
for _, f := range fields {
if f == "AND" || f == "OR" || f == "NOT" {
continue
}
kept = append(kept, f)
}
return strings.Join(kept, " ")
} }
// CreateMessage inserts a new message and returns the assigned ID. // CreateMessage inserts a new message and returns the assigned ID.
+36
View File
@@ -720,6 +720,42 @@ func TestSearchMessages_HyphenatedQuery(t *testing.T) {
} }
} }
// OC-0002: sanitizeFTSQuery only filters characters, not FTS5's bareword
// boolean keywords (AND, OR, NOT). A query consisting of (or containing) one
// of those keywords in an operator position makes SQLite raise an
// "fts5: syntax error", which SearchMessages surfaces as an error (mapped by
// the service layer to a 500) instead of returning zero results.
func TestSearchMessages_BooleanKeywordQuery(t *testing.T) {
database := openMigratedMemory(t)
userID := seedUser(t, database, "booleanuser")
chID := seedChannel(t, database, "booleanch")
_, _ = database.CreateMessage(context.Background(), chID, userID, "hello world", nil)
for _, q := range []string{"AND", "OR", "NOT", "AND world", "hello AND"} {
if _, err := database.SearchMessages(context.Background(), q, nil, 10); err != nil {
t.Errorf("SearchMessages(%q): unexpected error: %v", q, err)
}
}
}
// Same root cause as TestSearchMessages_BooleanKeywordQuery, but through the
// SearchMessagesInChannels caller, which sanitizes with the same
// sanitizeFTSQuery helper.
func TestSearchMessagesInChannels_BooleanKeywordQuery(t *testing.T) {
database := openMigratedMemory(t)
userID := seedUser(t, database, "boolchanuser")
chID := seedChannel(t, database, "boolchanch")
_, _ = database.CreateMessage(context.Background(), chID, userID, "hello world", nil)
for _, q := range []string{"AND", "OR", "NOT", "AND world", "hello AND"} {
if _, err := database.SearchMessagesInChannels(context.Background(), q, []int64{chID}, 10); err != nil {
t.Errorf("SearchMessagesInChannels(%q): unexpected error: %v", q, err)
}
}
}
// ─── UpdateReadState ────────────────────────────────────────────────────────── // ─── UpdateReadState ──────────────────────────────────────────────────────────
func TestUpdateReadState_Upsert(t *testing.T) { func TestUpdateReadState_Upsert(t *testing.T) {
+10 -6
View File
@@ -93,20 +93,24 @@ UPDATE voice_states SET server_deafened = 1, deafened = 1 WHERE user_id = ? AND
-- name: ClearVoiceServerDeafen :execresult -- name: ClearVoiceServerDeafen :execresult
UPDATE voice_states SET server_deafened = 0 WHERE user_id = ? AND channel_id = ?; UPDATE voice_states SET server_deafened = 0 WHERE user_id = ? AND channel_id = ?;
-- Camera and screenshare share one voice_max_video budget: a channel capped -- Camera and screenshare share one voice_max_video budget, counted in
-- at N simultaneous video streams must not let a camera publish ignore -- STREAMS, not rows: a channel capped at N simultaneous video streams must
-- screenshare occupants (or vice versa), so both gates count the same -- not let a camera publish ignore screenshare occupants (or vice versa,
-- `camera = 1 OR screenshare = 1` slot usage (OC-0023). -- OC-0023), and a single user with both flags set must consume two of the N
-- slots, not one (OC-0006) -- so both gates sum `vs2.camera + vs2.screenshare`
-- across the channel's rows rather than counting rows where either is set.
-- The enabling user's own bit is still 0 at gate time, so no self-exclusion
-- term is needed.
-- name: EnableCameraIfUnderLimit :execresult -- name: EnableCameraIfUnderLimit :execresult
UPDATE voice_states SET camera = 1 UPDATE voice_states SET camera = 1
WHERE voice_states.user_id = ? AND voice_states.channel_id = ? WHERE voice_states.user_id = ? AND voice_states.channel_id = ?
AND (SELECT COUNT(*) FROM voice_states AS vs2 WHERE vs2.channel_id = ? AND (vs2.camera = 1 OR vs2.screenshare = 1)) < ?; AND (SELECT COALESCE(SUM(vs2.camera), 0) + COALESCE(SUM(vs2.screenshare), 0) FROM voice_states AS vs2 WHERE vs2.channel_id = ?) < sqlc.arg(max_video);
-- name: EnableScreenshareIfUnderLimit :execresult -- name: EnableScreenshareIfUnderLimit :execresult
UPDATE voice_states SET screenshare = 1 UPDATE voice_states SET screenshare = 1
WHERE voice_states.user_id = ? AND voice_states.channel_id = ? WHERE voice_states.user_id = ? AND voice_states.channel_id = ?
AND (SELECT COUNT(*) FROM voice_states AS vs2 WHERE vs2.channel_id = ? AND (vs2.camera = 1 OR vs2.screenshare = 1)) < ?; AND (SELECT COALESCE(SUM(vs2.camera), 0) + COALESCE(SUM(vs2.screenshare), 0) FROM voice_states AS vs2 WHERE vs2.channel_id = ?) < sqlc.arg(max_video);
-- name: ClearVoiceState :exec -- name: ClearVoiceState :exec
DELETE FROM voice_states WHERE user_id = ?; DELETE FROM voice_states WHERE user_id = ?;
+2 -2
View File
@@ -278,7 +278,7 @@ func (d *DB) EnableCameraIfUnderLimit(ctx context.Context, userID, channelID int
UserID: userID, UserID: userID,
ChannelID: channelID, ChannelID: channelID,
ChannelID_2: channelID, ChannelID_2: channelID,
ChannelID_3: int64(maxVideo), MaxVideo: int64(maxVideo),
}) })
if err != nil { if err != nil {
return false, fmt.Errorf("EnableCameraIfUnderLimit: %w", err) return false, fmt.Errorf("EnableCameraIfUnderLimit: %w", err)
@@ -311,7 +311,7 @@ func (d *DB) EnableScreenshareIfUnderLimit(ctx context.Context, userID, channelI
UserID: userID, UserID: userID,
ChannelID: channelID, ChannelID: channelID,
ChannelID_2: channelID, ChannelID_2: channelID,
ChannelID_3: int64(maxVideo), MaxVideo: int64(maxVideo),
}) })
if err != nil { if err != nil {
return false, fmt.Errorf("EnableScreenshareIfUnderLimit: %w", err) return false, fmt.Errorf("EnableScreenshareIfUnderLimit: %w", err)
+13 -2
View File
@@ -273,9 +273,20 @@ func (s *DMService) CreateGroupDM(ctx context.Context, userID int64, recipientID
return nil, fmt.Errorf("%w: failed to create group DM", ErrInternal) return nil, fmt.Errorf("%w: failed to create group DM", ErrInternal)
} }
participants, err := s.st.GetDMParticipants(ctx, ch.ID, userID) // The channel, all dm_participants rows and all dm_open_state rows are
// already committed at this point, so this read must not turn a fully-
// persisted group DM into a reported failure (OC-0004): it runs
// uncancellable (context.WithoutCancel) so a client disconnect landing
// right after the commit can't fail it via ctx.Err(), and any other
// failure is logged rather than propagated so the caller still gets a
// usable result to fan dm_channel_open out from. Group DMs are
// duplicate-by-design — there is no "the group for these users" to find —
// so telling the caller creation failed when it actually committed only
// invites a retry that creates a second, indistinguishable group.
participants, err := s.st.GetDMParticipants(context.WithoutCancel(ctx), ch.ID, userID)
if err != nil { if err != nil {
return nil, fmt.Errorf("%w: failed to read group DM participants: %v", ErrInternal, err) slog.Error("DMService.CreateGroupDM: failed to read participants after commit", "err", err, "channel_id", ch.ID)
participants = nil
} }
return &CreateGroupDMResult{ return &CreateGroupDMResult{
+60
View File
@@ -61,3 +61,63 @@ func TestDMService_CreateGroupDM_RefusesBannedRecipient(t *testing.T) {
t.Fatalf("CreateGroupDM with a banned recipient = %v, want ErrNotFound", err) t.Fatalf("CreateGroupDM with a banned recipient = %v, want ErrNotFound", err)
} }
} }
// cancelAfterCreateGroupDMStore wraps a real *db.DB and cancels a context the
// instant CreateGroupDMChannel returns successfully — simulating a client
// disconnect that lands exactly in the gap between the channel's commit and
// the service's post-commit GetDMParticipants read.
type cancelAfterCreateGroupDMStore struct {
*db.DB
cancel context.CancelFunc
}
func (s *cancelAfterCreateGroupDMStore) CreateGroupDMChannel(ctx context.Context, name string, participantIDs []int64) (*db.Channel, error) {
ch, err := s.DB.CreateGroupDMChannel(ctx, name, participantIDs)
if err == nil {
s.cancel()
}
return ch, err
}
// OC-0004: CreateGroupDMChannel commits the channel, all dm_participants rows
// and all dm_open_state rows in one transaction. The subsequent
// GetDMParticipants read used to run on the same cancellable request context,
// so a client disconnect landing right after the commit (context cancelled
// in the gap) turned a fully-persisted group DM into a reported failure —
// inviting a client retry that, because group DMs are duplicate-by-design
// (db/dm_queries.go CreateGroupDMChannel doc), creates a second identical
// group.
func TestDMService_CreateGroupDM_SurvivesCancelledPostCommitRead(t *testing.T) {
database := newTestDB(t)
seedUser(t, database, &db.User{ID: 1, Username: "alice"})
seedUser(t, database, &db.User{ID: 2, Username: "bob"})
seedUser(t, database, &db.User{ID: 3, Username: "carol"})
ctx, cancel := context.WithCancel(context.Background())
st := &cancelAfterCreateGroupDMStore{DB: database}
st.cancel = cancel
svc := NewDMService(st)
result, err := svc.CreateGroupDM(ctx, 1, []int64{2, 3}, "")
if err != nil {
t.Fatalf("CreateGroupDM with context cancelled right after commit: %v (the channel is already persisted at this point — this must not fail the request)", err)
}
if result.Channel == nil {
t.Fatal("expected a created channel even though the post-commit context was cancelled")
}
if len(result.ParticipantIDs) != 3 {
t.Fatalf("ParticipantIDs = %v, want 3 entries so the caller can still broadcast dm_channel_open", result.ParticipantIDs)
}
// The channel must actually be persisted — a retry after this "failure"
// would otherwise be indistinguishable from creating a brand new group.
var count int
if err := database.QueryRowContext(context.Background(),
`SELECT COUNT(*) FROM dm_participants WHERE channel_id = ?`, result.Channel.ID,
).Scan(&count); err != nil {
t.Fatalf("count participants: %v", err)
}
if count != 3 {
t.Fatalf("persisted participant rows = %d, want 3", count)
}
}
+36 -9
View File
@@ -32,12 +32,33 @@ func (h *Hub) EmitEvents(ctx context.Context, events []Event) {
// An invisible user's public presence half rides this branch; // An invisible user's public presence half rides this branch;
// like the visible case below, it must invalidate any queued // like the visible case below, it must invalidate any queued
// coalescer entry so a stale connect-time presence can't flush // coalescer entry so a stale connect-time presence can't flush
// after (and overwrite) this fresher user-chosen status. // after (and overwrite) this fresher user-chosen status. The
// drop and the broadcast run atomically under presenceMu (see
// dropQueuedPresenceAndBroadcast, OC-0005) so a flush racing in
// at the same moment can never enqueue its stale frame after
// this fresher one.
if po, isPresence := ev.(PresenceOthersEvent); isPresence { if po, isPresence := ev.(PresenceOthersEvent); isPresence {
h.dropQueuedPresence(po.excludeUserID) h.dropQueuedPresenceAndBroadcast(po.excludeUserID, func() {
// Normal priority, excluding the owner — NOT
// broadcastExcludeLow. Every other source of this same
// user's presence (connect/disconnect via BroadcastToAll,
// and the visible presence_update path below) already
// shares the normal-priority queue; putting this one on
// the low-priority queue instead split one user's
// presence across two per-client FIFOs with different
// durability (low silently drops on overflow instead of
// disconnecting, so no replay ever repairs the loss) and
// different drain order (writePump always drains normal
// strictly before low, so a newer frame on one queue can
// be delivered before an older frame still sitting on the
// other) — exactly the hazard OC-0214 fixed for the
// visible case below (OC-0003).
h.BroadcastToAllExcept(po.excludeUserID, e.Payload())
})
} else {
// Low priority: typing indicators are ephemeral.
h.broadcastExcludeLow(e.ChannelID(), e.ExcludeUserID(), e.Payload())
} }
// Low priority: typing indicators are ephemeral.
h.broadcastExcludeLow(e.ChannelID(), e.ExcludeUserID(), e.Payload())
case UserTargetedEvent: case UserTargetedEvent:
// High priority: targeted events (DM opens, mentions). // High priority: targeted events (DM opens, mentions).
// dm_channel_open is unsequenced and targeted, so replay can never // dm_channel_open is unsequenced and targeted, so replay can never
@@ -75,12 +96,18 @@ func (h *Hub) EmitEvents(ctx context.Context, events []Event) {
// single ordered, seq-stamped, replayable stream (OC-0214). // single ordered, seq-stamped, replayable stream (OC-0214).
if pe, isPresence := ev.(PresenceEvent); isPresence { if pe, isPresence := ev.(PresenceEvent); isPresence {
// A user-chosen presence also bypasses the connect/disconnect // A user-chosen presence also bypasses the connect/disconnect
// coalescer; drop any entry still queued for this user or the // coalescer; drop any entry still queued for this user and
// pending flush (up to 300ms later) would overwrite this // broadcast atomically under presenceMu (see
// fresher status with the stale connect-time one. // dropQueuedPresenceAndBroadcast, OC-0005), or the pending
h.dropQueuedPresence(pe.userID) // flush (up to 300ms later) could race in between the drop
// and the broadcast and overwrite this fresher status with
// the stale connect-time one.
h.dropQueuedPresenceAndBroadcast(pe.userID, func() {
h.BroadcastToAll(e.Payload())
})
} else {
h.BroadcastToAll(e.Payload())
} }
h.BroadcastToAll(e.Payload())
default: default:
slog.Warn("EmitEvents: unknown event type", "type", fmt.Sprintf("%T", ev)) slog.Warn("EmitEvents: unknown event type", "type", fmt.Sprintf("%T", ev))
} }
@@ -0,0 +1,96 @@
package ws
// emit_presence_others_priority_test.go — regression test for OC-0003: the
// public half of an invisible user's presence (PresenceOthersEvent) went out
// via h.broadcastExcludeLow onto the low-priority queue — the same ephemeral,
// drop-on-overflow, unsequenced transport used for typing indicators — while
// every other source of the same user's presence (connect/disconnect via
// BroadcastToAll, and the visible presence_update path OC-0214 already fixed)
// shares the normal-priority queue. That split one user's presence across two
// per-client FIFOs with different durability and different drain order:
// writePump always drains normal strictly before low, so an observer with an
// older frame on low and a newer one on normal (or vice versa) can end up
// with the wrong one landing last, and a full low-priority queue silently
// drops the frame with no seq bump and no replay recovery — unlike the
// normal-priority queue, which disconnects on overflow and lets replay (or a
// fresh ready) repair the gap.
import (
"context"
"testing"
"time"
)
// TestEmitEvents_PresenceOthersEvent_UsesNormalPriorityQueue pins the fix: a
// PresenceOthersEvent (the public half of an invisible presence change),
// routed through the ExcludeSenderEvent case in EmitEvents, must land on an
// observer's normal-priority queue — the same FIFO every other presence
// source for that user uses — never the low-priority queue, while still never
// reaching the excluded user (the invisible owner) themselves.
//
// Before the fix, emit.go special-cased this case onto h.broadcastExcludeLow,
// so this test observes the frame on the observer's c.sendLow instead of
// c.send, and fails.
func TestEmitEvents_PresenceOthersEvent_UsesNormalPriorityQueue(t *testing.T) {
h := newEmitTestHub()
// Built directly (not via the emit_test.go helpers) so send and sendLow
// are DISTINCT channels — the shared-channel helpers in export_test.go
// unify them "for test observability" and would mask exactly the
// queue-split this test needs to detect.
observer := &Client{
hub: h,
ctx: context.Background(),
userID: 1,
send: make(chan []byte, 8),
sendHigh: make(chan []byte, 8),
sendLow: make(chan []byte, 8),
}
owner := &Client{
hub: h,
ctx: context.Background(),
userID: 2,
send: make(chan []byte, 8),
sendHigh: make(chan []byte, 8),
sendLow: make(chan []byte, 8),
}
h.clients[1] = observer
h.clients[2] = owner
h.pubsub.Subscribe(observer, TopicGlobal)
h.pubsub.Subscribe(owner, TopicGlobal)
// The normal-priority path goes through the async hub.broadcast channel,
// so the hub loop must be running to deliver it.
go h.Run()
defer h.Stop()
payload := []byte(`{"type":"presence","user_id":2,"status":"offline"}`)
h.EmitEvents(context.Background(), []Event{
PresenceOthersEvent{excludeUserID: 2, payload: payload},
})
observerNormal := drainChan(observer.send, 200*time.Millisecond)
observerLow := drainChan(observer.sendLow, 50*time.Millisecond)
ownerNormal := drainChan(owner.send, 50*time.Millisecond)
ownerLow := drainChan(owner.sendLow, 50*time.Millisecond)
if len(observerNormal) != 1 {
t.Errorf("expected the public half of an invisible presence change on the "+
"observer's normal-priority queue (same FIFO as connect/disconnect "+
"presence), got %d normal messages, %d low messages",
len(observerNormal), len(observerLow))
}
if len(observerLow) != 0 {
t.Errorf("invisible presence's public half must not go out on the "+
"low-priority queue: writePump drains normal strictly before low, so "+
"a frame queued there can be delivered after a later connect/"+
"disconnect presence frame on the normal queue, leaving the "+
"observer's final view stale, and is silently dropped (no replay "+
"recovery) on overflow; got %d low messages", len(observerLow))
}
if len(ownerNormal) != 0 || len(ownerLow) != 0 {
t.Errorf("the excluded owner must never receive the public half of "+
"their own invisible presence change: got %d normal, %d low messages",
len(ownerNormal), len(ownerLow))
}
}
+6 -2
View File
@@ -247,8 +247,12 @@ func (e PresenceEvent) Payload() []byte { return e.payload }
// PresenceOthersEvent is the public half of an invisible user's presence: the // PresenceOthersEvent is the public half of an invisible user's presence: the
// mapped ("offline") payload, broadcast to everyone except the user it // mapped ("offline") payload, broadcast to everyone except the user it
// describes. Satisfies ExcludeSenderEvent with a channel id of 0, which // describes. Satisfies ExcludeSenderEvent with a channel id of 0; EmitEvents
// broadcastExcludeLow routes as a global publish minus one subscriber. // special-cases it onto h.BroadcastToAllExcept (normal priority, sequenced,
// replayable) rather than the ExcludeSenderEvent default of
// broadcastExcludeLow, so this frame shares the same durable per-client FIFO
// as every other source of the same user's presence instead of the
// ephemeral, drop-on-overflow one built for typing indicators (OC-0003).
type PresenceOthersEvent struct { type PresenceOthersEvent struct {
excludeUserID int64 excludeUserID int64
payload []byte payload []byte
+100 -14
View File
@@ -20,6 +20,13 @@ type broadcastMsg struct {
// recipient's role may not READ, and the audience is resolved off the hub // recipient's role may not READ, and the audience is resolved off the hub
// goroutine so deliverBroadcast stays free of permission queries. // goroutine so deliverBroadcast stays free of permission queries.
recipients []int64 recipients []int64
// excludeUserID, when non-zero, is omitted from a global (channelID == 0)
// broadcast's live delivery. Used for the public half of an invisible
// user's presence (see BroadcastToAllExcept): everyone else must see it,
// but the owner's own view comes from a separate, synchronous, targeted
// send, and the two racing would let the async global broadcast overwrite
// it. Ignored outside the channelID == 0 branch of deliverBroadcast.
excludeUserID int64
// enqueuedAt stamps the enqueue site so deliverBroadcast can record // enqueuedAt stamps the enqueue site so deliverBroadcast can record
// enqueue→fanout latency. Zero on test-constructed messages; skipped then. // enqueue→fanout latency. Zero on test-constructed messages; skipped then.
enqueuedAt time.Time enqueuedAt time.Time
@@ -50,6 +57,29 @@ func (h *Hub) BroadcastToAll(msg []byte) {
} }
} }
// BroadcastToAllExcept enqueues msg for delivery to every connected client
// except excludeUserID. Non-blocking, like BroadcastToAll: if the broadcast
// channel is full the message is dropped with a warning.
//
// Routes through the SAME h.broadcast channel — and so the same single-
// goroutine hub dispatch loop and seqMu-serialized deliverBroadcast — as
// BroadcastToAll and every other normal-priority global broadcast. That
// shared serialization is what gives two broadcasts about the same user (say,
// a connect/disconnect presence frame and this one) their correct relative
// order at each observer: whichever enqueues onto h.broadcast first is also
// delivered to c.send first. A caller that instead published straight to
// pub/sub, bypassing this queue, would reintroduce exactly that kind of
// reordering from the other direction (OC-0003).
func (h *Hub) BroadcastToAllExcept(excludeUserID int64, msg []byte) {
select {
case h.broadcast <- broadcastMsg{channelID: 0, excludeUserID: excludeUserID, msg: msg, enqueuedAt: time.Now()}:
default:
h.broadcastDrops.Add(1)
slog.Warn("hub: broadcast channel full, dropping global message",
"msg_len", len(msg))
}
}
// broadcastVoiceEvent enqueues a voice_state / voice_leave message for the // broadcastVoiceEvent enqueues a voice_state / voice_leave message for the
// connected clients whose current role may READ channelID. // connected clients whose current role may READ channelID.
// //
@@ -599,27 +629,71 @@ func (h *Hub) QueuePresence(userID int64, status string, customStatus *string) {
} }
} }
// dropQueuedPresence removes a user's pending coalesced presence, if any. // dropQueuedPresenceAndBroadcast atomically removes any coalesced presence
// Called when a fresher presence for that user is broadcast directly (the // still queued for userID and runs broadcast, both under presenceMu. Called
// presence_update handler path), so the coalescer's later flush cannot // when a fresher presence for that user is delivered directly (the
// resurrect the stale connect-time state over it. Ordering holds because a // presence_update handler path, via EmitEvents), so the delete and the send
// user's connect (which queues) and their presence_update (which drops) run // of the fresher frame can never straddle flushPresenceQueue's own
// serially on the same connection's readPump. // snapshot-and-broadcast critical section (OC-0005).
func (h *Hub) dropQueuedPresence(userID int64) { //
// Holding presenceMu across the delete AND the broadcast — rather than just
// the delete — is what actually closes the race: whichever of this call and
// flushPresenceQueue acquires presenceMu second also enqueues its broadcast
// second.
// - If this call goes first, it deletes the entry before flush can ever
// snapshot it, so flush never broadcasts the stale state at all.
// - If flush goes first, this call's delete is a no-op against the
// already-cleared queue, but its broadcast still cannot run until flush's
// own broadcast has already been enqueued — so the fresher frame is
// stamped with the higher seq by deliverBroadcast's single FIFO consumer
// and every client's final view converges on it, not the stale one.
//
// broadcast runs with presenceMu held: every current caller (BroadcastToAll,
// BroadcastToAllExcept) only enqueues onto h.broadcast's non-blocking
// channel send, so this cannot block and introduces no new lock-order edge.
// Both callers sharing that same channel also means the "enqueues second"
// ordering guarantee above translates directly into delivery order: both
// broadcasts are drained by the same single-consumer hub dispatch loop
// (deliverBroadcast), in the order they were enqueued.
func (h *Hub) dropQueuedPresenceAndBroadcast(userID int64, broadcast func()) {
h.presenceMu.Lock() h.presenceMu.Lock()
defer h.presenceMu.Unlock()
delete(h.presenceQueue, userID) delete(h.presenceQueue, userID)
h.presenceMu.Unlock() broadcast()
} }
// presenceFlushRaceHook, when non-nil, runs once per flushPresenceQueue call
// immediately after the coalesced queue has been snapshotted and cleared,
// while presenceMu is still held. Test-only (always nil in production): the
// snapshot-to-broadcast window is too narrow to land a real concurrent
// dropQueuedPresenceAndBroadcast reliably, so tests use this hook to
// reproduce that interleaving deterministically. Mirrors the established
// refreshChannelVisibilityRaceHook / voiceJoinPostTokenRaceHook pattern.
var presenceFlushRaceHook func()
// flushPresenceQueue drains the coalescer and broadcasts each user's latest // flushPresenceQueue drains the coalescer and broadcasts each user's latest
// presence. Runs on the AfterFunc timer goroutine, never under presenceMu // presence, all under presenceMu (OC-0005). Runs on the AfterFunc timer
// during the fan-out. // goroutine.
//
// presenceMu is held across the broadcast loop, not just the snapshot: it
// used to be released beforehand, which let a concurrent
// dropQueuedPresenceAndBroadcast (nee dropQueuedPresence) call race in after
// the snapshot had already escaped the lock. The drop was then a guaranteed
// no-op against the live (already-nilled) map, AND nothing constrained
// whether that call's own fresher broadcast landed on h.broadcast before or
// after this loop's stale one — so the stale connect-time presence could win
// the seq race and permanently overwrite a user-chosen status. Holding the
// lock here forces the two critical sections to serialize, which is what
// dropQueuedPresenceAndBroadcast's ordering guarantee depends on.
func (h *Hub) flushPresenceQueue() { func (h *Hub) flushPresenceQueue() {
h.presenceMu.Lock() h.presenceMu.Lock()
defer h.presenceMu.Unlock()
queued := h.presenceQueue queued := h.presenceQueue
h.presenceQueue = nil h.presenceQueue = nil
h.presenceFlushArmed = false h.presenceFlushArmed = false
h.presenceMu.Unlock() if presenceFlushRaceHook != nil {
presenceFlushRaceHook()
}
for uid, p := range queued { for uid, p := range queued {
h.BroadcastPresence(uid, p.status, p.customStatus) h.BroadcastPresence(uid, p.status, p.customStatus)
} }
@@ -643,7 +717,15 @@ func (h *Hub) BroadcastPresence(userID int64, status string, customStatus *strin
// presencePayload.CustomStatus has no omitempty) so the client clears any // presencePayload.CustomStatus has no omitempty) so the client clears any
// cached text, matching what db.MemberSummary.ForViewer already does for // cached text, matching what db.MemberSummary.ForViewer already does for
// the ready payload's member list. // the ready payload's member list.
h.broadcastExcludeLow(0, userID, buildPresenceMsg(userID, public, nil)) //
// Normal priority, excluding the owner (BroadcastToAllExcept), not
// broadcastExcludeLow: the low-priority queue is unsequenced and dropped
// (not disconnected) on overflow, so it could silently lose this frame
// with no replay recovery, and — since writePump always drains normal
// strictly before low — deliver it out of order against the very
// connect/disconnect presence frames this same coalescer flush also
// produces for other users via BroadcastToAll (OC-0003).
h.BroadcastToAllExcept(userID, buildPresenceMsg(userID, public, nil))
h.SendToUser(userID, buildPresenceMsg(userID, status, customStatus)) h.SendToUser(userID, buildPresenceMsg(userID, status, customStatus))
} }
@@ -858,8 +940,12 @@ func (h *Hub) deliverBroadcast(bm broadcastMsg) {
h.SendToUser(userID, msg) h.SendToUser(userID, msg)
} }
case bm.channelID == 0: case bm.channelID == 0:
// Global broadcast — deliver to every connected client. // Global broadcast — deliver to every connected client, minus
h.pubsub.PublishGlobal(msg) // excludeUserID when the caller set one (see BroadcastToAllExcept).
// Publish(TopicGlobal, msg, 0) is exactly PublishGlobal(msg) when
// excludeUserID is the zero value, so ordinary BroadcastToAll
// callers are unaffected.
h.pubsub.Publish(TopicGlobal, msg, bm.excludeUserID)
default: default:
// Channel-scoped broadcast — deliver to subscribers of the channel // Channel-scoped broadcast — deliver to subscribers of the channel
// topic. The rate limiter already passed above, before the seq // topic. The rate limiter already passed above, before the seq
@@ -0,0 +1,119 @@
package ws
// OC-0006: EnableCameraIfUnderLimit / EnableScreenshareIfUnderLimit gate on
// COUNT(*) FROM voice_states WHERE ... (camera = 1 OR screenshare = 1) < N —
// one row per *user*, not one unit per *stream*. camera and screenshare are
// independent columns on the same row, so a single user with both flags set
// consumes only one slot in the count while actually publishing two streams.
// A channel capped at N simultaneous video streams can therefore over-admit
// to 2N live streams while still refusing the next publisher, claiming the
// N-stream cap is reached when more than N streams are already live.
import (
"context"
"testing"
"github.com/owncord/server/db"
"github.com/owncord/server/permissions"
)
const oc0006VideoStreamRoleID = int64(211)
func newOC0006VideoStreamDB(t *testing.T) *db.DB {
t.Helper()
database, err := db.Open(":memory:")
if err != nil {
t.Fatalf("db.Open: %v", err)
}
if err := db.Migrate(database); err != nil {
t.Fatalf("Migrate: %v", err)
}
t.Cleanup(func() { _ = database.Close() })
if _, err := database.ExecContext(context.Background(),
`INSERT INTO roles (id, name, color, permissions, position, is_default)
VALUES (?, 'oc-0006-video', NULL, ?, 5, 0)`,
oc0006VideoStreamRoleID,
permissions.ReadMessages|permissions.ConnectVoice|permissions.SpeakVoice|permissions.UseVideo|permissions.ShareScreen,
); err != nil {
t.Fatalf("seed oc-0006-video role: %v", err)
}
return database
}
func seedOC0006VideoStreamUser(t *testing.T, database *db.DB, username string) int64 {
t.Helper()
uid, err := database.CreateUser(context.Background(), username, "hash", int(oc0006VideoStreamRoleID))
if err != nil {
t.Fatalf("CreateUser %s: %v", username, err)
}
return uid
}
// mustCreateVideoCappedChannel2 mirrors mustCreateVideoCappedChannel from
// oc_0023_screenshare_video_limit_test.go (kept file-local to avoid a
// cross-file test helper dependency).
func mustCreateVideoCappedChannel2(t *testing.T, database *db.DB, name string, maxVideo int) int64 {
t.Helper()
chID, err := database.CreateChannel(context.Background(), name, "voice", "", "", 0)
if err != nil {
t.Fatalf("CreateChannel %s: %v", name, err)
}
if err := database.AdminUpdateChannel(context.Background(), chID, db.ChannelUpdate{
Name: name,
VoiceMaxVideo: maxVideo,
}); err != nil {
t.Fatalf("AdminUpdateChannel %s: %v", name, err)
}
return chID
}
// A channel capped at 2 simultaneous video streams. Alice alone publishes
// both her camera and her screenshare -- that is 2 streams from one row.
// Bob's camera enable must then be refused: the cap is already saturated by
// Alice's two streams. Today the gate counts Alice's row once (camera=1 OR
// screenshare=1 matches her single row), so it reads slot usage as 1, not 2,
// and wrongly admits Bob's third stream.
func TestEnableVideoSlot_SameUserDoubleStreamCountsTwoSlots(t *testing.T) {
ctx := context.Background()
database := newOC0006VideoStreamDB(t)
chID := mustCreateVideoCappedChannel2(t, database, "capped-room-samerow", 2)
alice := seedOC0006VideoStreamUser(t, database, "alice-double")
bob := seedOC0006VideoStreamUser(t, database, "bob-third")
if err := database.JoinVoiceChannel(ctx, alice, chID); err != nil {
t.Fatalf("JoinVoiceChannel alice: %v", err)
}
if err := database.JoinVoiceChannel(ctx, bob, chID); err != nil {
t.Fatalf("JoinVoiceChannel bob: %v", err)
}
d := VoiceDeps{DB: database, Permissions: permissions.NewChecker(database)}
ssRes := handleVoiceScreenshareV2(ctx, VoiceScreenshareCmd{userID: alice, enabled: true}, ClientInfo{UserID: alice, VoiceChannelID: chID}, d)
if ssRes.Error != nil {
t.Fatalf("alice screenshare enable (1st stream, cap 2) should succeed, got error: %+v", ssRes.Error)
}
camRes := handleVoiceCameraV2(ctx, VoiceCameraCmd{userID: alice, enabled: true}, ClientInfo{UserID: alice, VoiceChannelID: chID}, d)
if camRes.Error != nil {
t.Fatalf("alice camera enable (2nd stream, cap 2) should succeed, got error: %+v", camRes.Error)
}
// Cap is now saturated: Alice alone is publishing 2 of the 2 allowed
// streams. Bob's camera enable is a 3rd stream and must be refused.
bobCamRes := handleVoiceCameraV2(ctx, VoiceCameraCmd{userID: bob, enabled: true}, ClientInfo{UserID: bob, VoiceChannelID: chID}, d)
if bobCamRes.Error == nil {
t.Fatal("bob's camera enable succeeded as the channel's 3rd live video stream against a cap of 2 -- the same-user double-publish (camera+screenshare on one row) was undercounted as a single slot")
}
if ce, ok := bobCamRes.Error.(ClientError); !ok || ce.Code != ErrCodeVideoLimit {
t.Errorf("error = %+v, want ClientError{Code: %q}", bobCamRes.Error, ErrCodeVideoLimit)
}
vs, err := database.GetVoiceState(ctx, bob)
if err != nil || vs == nil {
t.Fatalf("GetVoiceState bob: %v", err)
}
if vs.Camera {
t.Error("bob's camera flag was set to true despite the VIDEO_LIMIT refusal")
}
}
@@ -0,0 +1,97 @@
package ws
// presence_coalesce_flush_race_test.go — regression test for OC-0005.
//
// flushPresenceQueue snapshots h.presenceQueue and releases presenceMu
// BEFORE broadcasting the snapshotted entries. dropQueuedPresence (the guard
// EmitEvents uses to stop a stale connect/disconnect presence from
// clobbering a fresher user-chosen status) only deletes from the LIVE map,
// so a drop that lands after the flush has already taken its snapshot is a
// no-op — and, critically, nothing then constrains the relative order in
// which the flush's stale broadcast and the fresher direct broadcast reach
// h.broadcast. Both go through deliverBroadcast's single FIFO consumer,
// which stamps seq in enqueue order, so whichever one is enqueued LAST wins
// every client's final view. A stale connect-time presence enqueued after a
// user's own fresher presence_update therefore permanently overwrites it.
//
// The snapshot-to-broadcast window is a few instructions wide and not
// reliably landed by staggering real goroutines, so presenceFlushRaceHook
// (test-only, nil in production) fires at exactly that point, mirroring the
// established refreshChannelVisibilityRaceHook / voiceJoinPostTokenRaceHook
// pattern used to pin analogous races elsewhere in this package.
import (
"bytes"
"context"
"testing"
"time"
)
// TestFlushPresenceQueue_ConcurrentDirectPresenceOrdersLast pins OC-0005 for
// the visible presence_update path (EmitEvents' BroadcastAllEvent branch).
//
// A stale connect-time "online" is queued for user 42. While
// flushPresenceQueue is mid-flush (queue already snapshotted), a concurrent
// presence_update to "dnd" races in via EmitEvents. The fresher "dnd" must
// end up enqueued on h.broadcast AFTER the stale "online" — so the hub
// stamps it with the higher seq and every other client's final view of user
// 42 converges on "dnd", not the stale "online".
func TestFlushPresenceQueue_ConcurrentDirectPresenceOrdersLast(t *testing.T) {
h := &Hub{
broadcast: make(chan broadcastMsg, 8),
pubsub: NewPubSub(),
}
// Populate the queue directly rather than via QueuePresence: QueuePresence
// arms a real 300ms time.AfterFunc(h.flushPresenceQueue), and this test
// already drives flushPresenceQueue manually below. Leaving that timer
// armed would let it fire later — during a *later* run of this same test
// under -count=N, or during another test entirely — and invoke whatever
// presenceFlushRaceHook happens to be installed at that later moment,
// which is exactly the kind of cross-run interference this test must not
// introduce.
h.presenceMu.Lock()
h.presenceQueue = map[int64]pendingPresence{42: {status: "online"}}
h.presenceFlushArmed = true
h.presenceMu.Unlock()
raced := make(chan struct{})
var hookRan bool
presenceFlushRaceHook = func() {
hookRan = true
// Simulate EmitEvents' direct presence branch racing in exactly
// here: after flushPresenceQueue has snapshotted (and, currently,
// released presenceMu for) the queue, but before it has broadcast
// the stale entry.
go func() {
h.EmitEvents(context.Background(), presenceEvents(42, "dnd", nil))
close(raced)
}()
// Give the goroutine room to actually run: with the bug this lets
// it complete its (unsynchronized) broadcast well before flush's
// own loop runs; with the fix in place the goroutine instead blocks
// on presenceMu until this function returns and releases it, so the
// sleep costs nothing extra there either way.
time.Sleep(20 * time.Millisecond)
}
defer func() { presenceFlushRaceHook = nil }()
h.flushPresenceQueue()
<-raced
if !hookRan {
t.Fatal("presenceFlushRaceHook never fired — test setup is broken, not exercising the flush race window")
}
if len(h.broadcast) != 2 {
t.Fatalf("expected 2 broadcast frames (stale flush + fresh update), got %d", len(h.broadcast))
}
first := <-h.broadcast
second := <-h.broadcast
if !bytes.Contains(first.msg, []byte(`"status":"online"`)) {
t.Errorf("expected the stale flush's 'online' frame enqueued FIRST; got first=%s", first.msg)
}
if !bytes.Contains(second.msg, []byte(`"status":"dnd"`)) {
t.Errorf("expected the fresh presence_update's 'dnd' frame enqueued LAST (so it gets the higher seq and wins); got second=%s", second.msg)
}
}
@@ -0,0 +1,184 @@
package ws
// reconnect_fallback_channel_leak_test.go — regression test for OC-0001.
//
// handleReconnect promotes an attacker-supplied (but READ-gated at the time)
// active_channel_id into c.channelID (serve.go ~349) before its own abort
// paths run. When a permission revocation lands deep in the handshake and
// trips the final mustFullResync re-check (serve.go ~401), handleReconnect
// aborts with handled=false and ServeWS falls through to handleFreshConnect
// — but nothing clears the already-set c.channelID. handleFreshConnect
// recomputes allowedChannelIDs WITHOUT the revoked channel and its
// buildReady payload correctly omits it, but registerNow (hub.go ~560)
// subscribes c.channelID's ChannelTopic unconditionally, with no
// readableChannelIDs check — unlike its two siblings in the same function.
// Every subsequent broadcast to that channel is then delivered to a client
// that was never granted READ_MESSAGES for it.
import (
"context"
"encoding/json"
"net/http/httptest"
"strings"
"testing"
"time"
"github.com/coder/websocket"
"github.com/owncord/server/auth"
"github.com/owncord/server/permissions"
)
func TestReconnect_FullReadyFallbackDoesNotLeakRevokedChannelSubscription(t *testing.T) {
database := newHarvestVoiceDB(t)
ctx := context.Background()
uid := seedHarvestVoiceUser(t, database, "fallback-leak-user")
chID := mustCreateVoiceChannel(t, database, "fallback-leak-channel")
ch, err := database.GetChannel(ctx, chID)
if err != nil || ch == nil {
t.Fatalf("GetChannel: %v", err)
}
user, err := database.GetUserByID(ctx, uid)
if err != nil || user == nil {
t.Fatalf("GetUserByID: %v", err)
}
token, err := auth.GenerateToken()
if err != nil {
t.Fatalf("GenerateToken: %v", err)
}
if _, err := database.CreateSession(ctx, uid, auth.HashToken(token), "test", "127.0.0.1"); err != nil {
t.Fatalf("CreateSession: %v", err)
}
hub := NewHub(database, auth.NewRateLimiter(), nil)
go hub.Run()
defer hub.Stop()
// Precondition: the channel is READ-visible at the moment the auth frame
// is evaluated, so the auth-frame active_channel_id is legitimately
// honoured by handleReconnect.
allowedBefore, err := hub.computeAllowedChannels(ctx, database, user)
if err != nil {
t.Fatalf("computeAllowedChannels: %v", err)
}
if !allowedBefore[chID] {
t.Fatalf("precondition: channel %d should start READ-visible", chID)
}
// Bracket last_seq=99 so the resume takes the buffer tier (not a
// mustFullResync-forced full ready from the very start).
rb := hub.ReplayBuffer()
rb.Push(98, chID, []byte(`{"seq":98,"type":"chat_message","payload":{}}`))
rb.Push(99, chID, []byte(`{"seq":99,"type":"chat_message","payload":{}}`))
rb.Push(100, chID, []byte(`{"seq":100,"type":"chat_message","payload":{}}`))
hub.SeedSeq(100)
const lastSeq = uint64(99)
if hub.mustFullResync(lastSeq) {
t.Fatalf("precondition: mustFullResync must be false before any visibility change")
}
// Fires once, deep inside handleReconnect — after c.channelID has already
// been promoted from active_channel_id (serve.go ~349) but before the
// final mustFullResync re-check (serve.go ~401). Revoke READ_MESSAGES on
// chID, mirroring an admin edit racing the resume, exactly like
// TestHandleReconnect_VisibilityChangeDuringHandshake_ForcesFullReady.
var hookRan bool
handleReconnectPreRegisterRaceHook = func() {
hookRan = true
if overrideErr := database.UpsertChannelOverride(ctx, chID, harvestVoiceRoleID, 0, permissions.ReadMessages); overrideErr != nil {
t.Fatalf("UpsertChannelOverride: %v", overrideErr)
}
//nolint:contextcheck // RefreshChannelVisibility takes no context by design.
hub.RefreshChannelVisibility(ch)
}
defer func() { handleReconnectPreRegisterRaceHook = nil }()
srv := httptest.NewServer(ServeWS(hub, database, []string{"*"}, 0))
defer srv.Close()
dialCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
defer cancel()
conn, dialResp, dialErr := websocket.Dial(dialCtx, "ws"+strings.TrimPrefix(srv.URL, "http"), nil)
if dialResp != nil && dialResp.Body != nil {
_ = dialResp.Body.Close()
}
if dialErr != nil {
t.Fatalf("websocket.Dial: %v", dialErr)
}
defer func() { _ = conn.Close(websocket.StatusNormalClosure, "") }()
raw, _ := json.Marshal(map[string]any{
"type": "auth",
"payload": map[string]any{
"token": token,
"last_seq": lastSeq,
"active_channel_id": chID,
},
})
if err := conn.Write(dialCtx, websocket.MessageText, raw); err != nil {
t.Fatalf("write auth: %v", err)
}
// The abort forces the fallback full-connect flow, which writes TWO
// frames — auth_ok, then ready — unlike the single auth_ok a successful
// buffer-tier resume would send.
readCtx, readCancel := context.WithTimeout(ctx, 5*time.Second)
defer readCancel()
_, authMsg, err := conn.Read(readCtx)
if err != nil {
t.Fatalf("read auth_ok: %v", err)
}
var authParsed map[string]any
if err := json.Unmarshal(authMsg, &authParsed); err != nil {
t.Fatalf("unmarshal auth_ok: %v", err)
}
if authParsed["type"] != MsgTypeAuthOK {
t.Fatalf("expected auth_ok, got %v", authParsed["type"])
}
_, readyMsg, err := conn.Read(readCtx)
if err != nil {
t.Fatalf("read ready: %v", err)
}
var readyParsed map[string]any
if err := json.Unmarshal(readyMsg, &readyParsed); err != nil {
t.Fatalf("unmarshal ready: %v", err)
}
if readyParsed["type"] != MsgTypeReady {
t.Fatalf("expected ready (handleReconnect aborted, fell through to handleFreshConnect), got %v", readyParsed["type"])
}
if !hookRan {
t.Fatal("handleReconnectPreRegisterRaceHook never fired — test setup is broken, not exercising the race window")
}
deadline := time.Now().Add(2 * time.Second)
var c *Client
for {
hub.mu.Lock()
c = hub.clients[uid]
hub.mu.Unlock()
if c != nil {
break
}
if time.Now().After(deadline) {
t.Fatal("client was never registered")
}
time.Sleep(10 * time.Millisecond)
}
if got := c.getChannelID(); got == chID {
t.Errorf("resumed-then-fallback client kept focus on revoked channel %d after registerNow, want 0", got)
}
hub.pubsub.mu.RLock()
sub := hub.pubsub.topics[ChannelTopic(chID)][uid]
hub.pubsub.mu.RUnlock()
if sub != nil {
t.Errorf("client is subscribed to ChannelTopic(%d) despite READ_MESSAGES being revoked before registration — "+
"every subsequent broadcast to that channel will be delivered to this socket", chID)
}
}
+14
View File
@@ -692,6 +692,20 @@ func (h *Hub) handleFreshConnect(
allowedChannelIDs = allowed allowedChannelIDs = allowed
} }
} }
// handleReconnect may have promoted an auth-frame active_channel_id into
// c.channelID (serve.go, honoured only when it was READ-visible at that
// moment) and then aborted on one of its own re-checks — most notably the
// final mustFullResync check, tripped by a permission revocation that
// landed mid-handshake. None of those abort paths undo the c.channelID
// write. registerNow subscribes c.channelID's ChannelTopic
// unconditionally, so re-gate it here against the freshly recomputed
// permission set before registering. Fail closed: a nil allowedChannelIDs
// (lastSeq == 0, or the computeAllowedChannels error branch above) denies.
if chID := c.getChannelID(); chID != 0 && !allowedChannelIDs[chID] {
c.mu.Lock()
c.channelID = 0
c.mu.Unlock()
}
h.registerNow(c, allowedChannelIDs) h.registerNow(c, allowedChannelIDs)
// Settle the session's status before buildReady reads the member list, so // Settle the session's status before buildReady reads the member list, so
+32
View File
@@ -75,6 +75,29 @@ func (h *Hub) presentableMembers(members []db.MemberSummary, viewerID int64) []d
return out return out
} }
// presentableDMChannels applies presentableMembers' "no live connection means
// offline" rule to a DM channel list's recipient statuses. GetUserDMChannels
// already applies db.StatusForViewer (the invisible-to-others half); this
// adds the missing "no live connection" half so dm_channels cannot disagree
// with the members array about whether the same disconnected user is online.
// Both Recipient (the legacy single-recipient field) and every entry of
// Recipients (the group-aware field) are rewritten, since a 1:1 DM's
// Recipient is a copy of Recipients[0], not a shared reference.
func (h *Hub) presentableDMChannels(dmChannels []db.DMChannelInfo) []db.DMChannelInfo {
connected := h.connectedUserIDs()
for i := range dmChannels {
if dmChannels[i].Recipient.ID != 0 && !connected[dmChannels[i].Recipient.ID] {
dmChannels[i].Recipient.Status = db.StatusOffline
}
for j := range dmChannels[i].Recipients {
if !connected[dmChannels[i].Recipients[j].ID] {
dmChannels[i].Recipients[j].Status = db.StatusOffline
}
}
}
return dmChannels
}
// connectedUserIDs snapshots the ids with a live WebSocket connection. // connectedUserIDs snapshots the ids with a live WebSocket connection.
func (h *Hub) connectedUserIDs() map[int64]bool { func (h *Hub) connectedUserIDs() map[int64]bool {
h.mu.RLock() h.mu.RLock()
@@ -249,6 +272,15 @@ func (h *Hub) buildReady(ctx context.Context, database *db.DB, userID int64, rol
dmChannels[i].MentionCount = u.MentionCount dmChannels[i].MentionCount = u.MentionCount
} }
} }
// GetUserDMChannels only applies db.StatusForViewer, which collapses
// invisible to offline but passes a disconnected recipient's saved
// idle/dnd through verbatim (MarkUserDisconnected deliberately keeps a
// chosen idle/dnd across a disconnect so the next connect can honour it,
// relying on every read path to hide it in the meantime). members already
// gets the "no live connection means offline" half of that rule from
// presentableMembers above; apply the same half here so dm_channels
// cannot disagree with members about the same user within one payload.
dmChannels = h.presentableDMChannels(dmChannels)
// Collect voice states, filtered to visible channels (BUG-095) plus the // Collect voice states, filtered to visible channels (BUG-095) plus the
// user's own open DM channels — mirroring computeAllowedChannels, which // user's own open DM channels — mirroring computeAllowedChannels, which
+102
View File
@@ -0,0 +1,102 @@
package ws_test
// serve_ready_dm_status_test.go — regression test for OC-0008: buildReady's
// dm_channels half came straight from database.GetUserDMChannels, which
// applies only db.StatusForViewer — that collapses invisible to offline but
// passes a disconnected user's saved idle/dnd through verbatim. members goes
// through presentableMembers first, which additionally forces offline for
// anyone with no live WebSocket connection (mirroring
// TestReady_DisconnectedMemberWithChosenStatusRendersOffline in
// presence_invisible_test.go, but for the dm_channels field). Before the fix
// a signed-out user who last chose "dnd" or "idle" would show as offline in
// members but still dnd/idle in dm_channels within the very same ready frame.
import (
"context"
"encoding/json"
"testing"
"github.com/owncord/server/db"
)
// dmChannelStatusFor pulls the recipient status for other.ID out of a ready
// payload's dm_channels array, checking both the legacy `recipient` field and
// the group-aware `recipients` array so a fix that only patches one leaks
// through undetected.
func dmChannelStatusFor(t *testing.T, raw []byte, otherID int64) (recipientStatus string, recipientsStatus string, found bool) {
t.Helper()
var env struct {
Payload struct {
DMChannels []struct {
Recipient struct {
ID int64 `json:"id"`
Status string `json:"status"`
} `json:"recipient"`
Recipients []struct {
ID int64 `json:"id"`
Status string `json:"status"`
} `json:"recipients"`
} `json:"dm_channels"`
} `json:"payload"`
}
if err := json.Unmarshal(raw, &env); err != nil {
t.Fatalf("unmarshal ready: %v", err)
}
for _, dm := range env.Payload.DMChannels {
if dm.Recipient.ID == otherID {
recipientStatus = dm.Recipient.Status
found = true
}
for _, r := range dm.Recipients {
if r.ID == otherID {
recipientsStatus = r.Status
}
}
}
return recipientStatus, recipientsStatus, found
}
// TestBuildReady_DMChannelsHidesDisconnectedRecipientStatus pins OC-0008: a DM
// recipient with no live WebSocket connection must render offline in
// dm_channels, exactly as presentableMembers already forces for the members
// array. absent chooses "dnd", then MarkUserDisconnected-equivalent state is
// simulated by simply never registering a client for absent (buildReady's
// connectedUserIDs() only reflects live hub registrations, so an
// unregistered user is indistinguishable from "signed out").
func TestBuildReady_DMChannelsHidesDisconnectedRecipientStatus(t *testing.T) {
hub, database := newServeHub(t)
ctx := context.Background()
viewer := seedServeUser(t, database, "dm-status-viewer")
absent := seedServeUser(t, database, "dm-status-absent")
viewerRole, err := database.GetRoleByID(ctx, viewer.RoleID)
if err != nil || viewerRole == nil {
t.Fatalf("GetRoleByID: %v", err)
}
// absent chose "dnd" before signing out; the column keeps it (this is
// exactly what MarkUserDisconnected leaves behind for a non-online status).
if err := database.UpdateUserStatus(ctx, absent.ID, db.StatusDND); err != nil {
t.Fatalf("UpdateUserStatus: %v", err)
}
seedDMChannel(t, database, viewer.ID, absent.ID)
// Only the viewer has a live connection; absent is never registered, so
// they must render offline everywhere in this ready payload.
msg, err := hub.BuildReadyWithRoleForTest(database, viewer.ID, viewerRole)
if err != nil {
t.Fatalf("BuildReadyWithRoleForTest: %v", err)
}
recipientStatus, recipientsStatus, found := dmChannelStatusFor(t, msg, absent.ID)
if !found {
t.Fatalf("ready payload's dm_channels is missing recipient %d", absent.ID)
}
if recipientStatus != db.StatusOffline {
t.Errorf("dm_channels[].recipient.status = %q, want %q (disconnected member must render offline, same rule presentableMembers applies to the members array)", recipientStatus, db.StatusOffline)
}
if recipientsStatus != db.StatusOffline {
t.Errorf("dm_channels[].recipients[].status = %q, want %q", recipientsStatus, db.StatusOffline)
}
}