aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--src/features/messages.js34
-rw-r--r--tools/go-live-test.js36
2 files changed, 58 insertions, 12 deletions
diff --git a/src/features/messages.js b/src/features/messages.js
index 4820baa..c82670b 100644
--- a/src/features/messages.js
+++ b/src/features/messages.js
@@ -8975,7 +8975,20 @@ export function messagesFeature(ctx) {
b.ws = ws;
bcRecord(b);
b.reconnecting = false;
- if (info.id !== b.id) toast(t('goLiveRestarted'));
+ // The server couldn't take us back (it restarted, or we were gone
+ // too long): this is a new stream at a new address. The old one is
+ // over — said so, with no recording (there's next to nothing in
+ // it) — and the new one is announced as soon as it plays.
+ if (info.id !== b.id) {
+ toast(t('goLiveRestarted'));
+ const old = { ...b, recording: null };
+ Object.assign(b, { id: info.id, liveUrl: info.live, recording: info.recording, a: LIVE_KIND + ':' + b.pk + ':' + info.id });
+ bcAnnounce(old, 'ended').catch(() => {});
+ bcPlaying(b).then(() => (bc === b && b.phase === 'live' ? bcAnnounce(b, 'live') : null)).then(() => {
+ // the streamer's own view of it follows to the new address
+ if (bc === b && b.ev && ui.liveStream && ui.liveStream.a === old.a) { ui.liveStream = { a: b.a, seed: b.ev }; render(); }
+ }).catch(() => {});
+ }
render();
return;
} catch {}
@@ -8994,14 +9007,23 @@ export function messagesFeature(ctx) {
['p', id.pubkey, '', 'Host'], ['relays', ...relays.slice(0, 6)], CLIENT_TAG];
if (me.picture) tags.push(['image', me.picture]);
if (status === 'live' && b.viewers != null) tags.push(['current_participants', String(b.viewers)]);
- if (status === 'ended') tags.push(['ends', String(Math.floor(Date.now() / 1000))], ['recording', b.recording]);
+ if (status === 'ended') tags.push(['ends', String(Math.floor(Date.now() / 1000))]);
+ if (status === 'ended' && b.recording) tags.push(['recording', b.recording]);
const partial = { kind: LIVE_KIND, content: '', created_at: Math.floor(Date.now() / 1000), tags };
const evt = id.signer instanceof Uint8Array ? finalizeEvent(partial, id.signer) : await id.signer.signEvent(partial);
takeLive([evt]);
- b.ev = evt;
+ if (bc === b) b.ev = evt;
scheduleRepaint();
return publishOn(relays, evt);
}
+ async function bcPlaying(b) {
+ const url = b.liveUrl;
+ for (let i = 0; i < 40 && bc === b && b.phase === 'live' && b.liveUrl === url; i++) {
+ await new Promise((r) => setTimeout(r, 500));
+ const pl = await fetch(url, { cache: 'no-store' }).then((r) => r.text()).catch(() => '');
+ if ((pl.match(/\.ts\s*$/gm) || []).length >= 3) break;
+ }
+ }
async function startBroadcast() {
const b = bc;
if (!b || b.phase !== 'setup' || !b.media) return;
@@ -9019,11 +9041,7 @@ export function messagesFeature(ctx) {
// three segments — Chrome's own player won't start a live playlist with
// fewer (it fails to parse it), so a viewer who comes at once would get
// an error, not the stream
- for (let i = 0; i < 40 && bc === b && b.phase === 'live'; i++) {
- await new Promise((r) => setTimeout(r, 500));
- const pl = await fetch(b.liveUrl, { cache: 'no-store' }).then((r) => r.text()).catch(() => '');
- if ((pl.match(/\.ts\s*$/gm) || []).length >= 3) break;
- }
+ await bcPlaying(b);
if (bc !== b || b.phase !== 'live') return;
await bcAnnounce(b, 'live');
b.tick = setInterval(() => bcTick(b), 60_000);
diff --git a/tools/go-live-test.js b/tools/go-live-test.js
index 82d6c7c..61e3c02 100644
--- a/tools/go-live-test.js
+++ b/tools/go-live-test.js
@@ -8,7 +8,7 @@
import puppeteer from './puppeteer.js';
import { cacheKeyFor } from '../src/wallet.js';
import { naddrEncode } from 'nostr-tools/nip19';
-import { mkdtempSync, rmSync } from 'node:fs';
+import { mkdtempSync, rmSync, readFileSync, writeFileSync } from 'node:fs';
import { tmpdir } from 'node:os';
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
let ok = true;
@@ -23,8 +23,17 @@ const names = Bun.serve({ port: 0, hostname: '127.0.0.1', fetch: (req) => Respon
process.env.LIVE_DATA = dir;
process.env.NAMES_URL = names.url.origin;
process.env.LIVE_BASE = `http://localhost:${PORT}/live`;
-const live = await import(process.env.LIVE_SERVER || '../../coinosv3-server/live/server.js');
-const server = Bun.serve({ port: PORT, websocket: live.websocket, fetch: async (req, srv) => {
+const LIVE_PATH = process.env.LIVE_SERVER || '../../coinosv3-server/live/server.js';
+let live = await import(LIVE_PATH);
+// each socket stays with the server (module) that took it, so a "restart"
+// (a fresh import) leaves the old connections to the old one
+const sockMod = new WeakMap();
+const websocket = {
+ open: (ws) => { sockMod.set(ws, live); live.websocket.open?.(ws); },
+ message: (ws, m) => (sockMod.get(ws) || live).websocket.message?.(ws, m),
+ close: (ws, c, r) => (sockMod.get(ws) || live).websocket.close?.(ws, c, r),
+};
+const server = Bun.serve({ port: PORT, websocket, fetch: async (req, srv) => {
const path = new URL(req.url).pathname;
if (path.startsWith('/live/')) {
if (path === '/live/ingest') try { owner = JSON.parse(atob(new URL(req.url).searchParams.get('auth'))).pubkey; } catch {}
@@ -43,7 +52,7 @@ const relayStub = (fx) => {
const Real = window.WebSocket;
window.WebSocket = function (...a) {
const ws = new Real(...a);
- if (/\/live\/ingest/.test(String(a[0]))) return ws; // the live server itself
+ if (/\/live\/ingest/.test(String(a[0]))) { (window.__ingest = window.__ingest || []).push(ws); return ws; } // the live server itself
const send = ws.send.bind(ws);
ws.send = (data) => {
try {
@@ -131,6 +140,25 @@ try {
check('the server counts the watcher', st.viewers >= 1, JSON.stringify(st));
await viewer.close();
+ console.log('\n[the server restarts mid-stream]');
+ // a fresh server, and the old one's shutdown: its streams end
+ live = await import(LIVE_PATH + '?restart=1');
+ const oldId = tag(ann, 'd');
+ const sj = `${dir}/${oldId}/stream.json`;
+ writeFileSync(sj, JSON.stringify({ ...JSON.parse(readFileSync(sj, 'utf8')), state: 'ended', ended: Date.now() }));
+ const before = (await lives(page)).length;
+ await page.evaluate(() => window.__ingest.forEach((w) => w.close()));
+ check('the app comes back as a new stream and announces it', await waitFor((n) => window.__published.filter((e) => e.kind === 30311).slice(n)
+ .some((e) => e.tags.some((x) => x[0] === 'status' && x[1] === 'live')), 40000, before));
+ const after = (await lives(page)).slice(before);
+ const again = after.find((e) => tag(e, 'status') === 'live');
+ check('...at its new address, not the dead one', again && tag(again, 'd') !== oldId && tag(again, 'streaming').includes(tag(again, 'd')), JSON.stringify(after.map((e) => [tag(e, 'd'), tag(e, 'status')])));
+ const gone = after.find((e) => tag(e, 'd') === oldId);
+ check('...the old one said to be over, with no recording', gone && tag(gone, 'status') === 'ended' && !gone.tags.some((x) => x[0] === 'recording'));
+ const pl = await (await fetch(tag(again, 'streaming'))).text();
+ check('...and the new one plays', (pl.match(/\.ts/g) || []).length >= 3, (pl.match(/\.ts/g) || []).length + ' segments');
+ check('the streamer\'s own page follows to it', await waitFor((a) => !!document.querySelector('.live-page .bc-bar'), 5000));
+
console.log('\n[the end]');
await page.evaluate(() => document.querySelector('.bc-end').click());
check('End leaves the stream\'s page', await waitFor(() => !document.querySelector('.bc-bar'), 5000));
generated by cgit v1.2.3 (git 2.51.0) at 2026-10-19 13:32:48 +0000