fix: make humanised selfbot startup abort- and concurrency-safe
The human-pause delays leave start() in-flight for several seconds, which
exposed two races:
- stop() during a pause only ended the pause; start() continued and called
joinVoice on the streamer stop() had already nulled (null deref).
- `active` was set only just before go-live, so a second /stream during the
delay passed the guard and both calls raced on the same overwritten streamer.
Now start() locks `active` before any await, keeps controller/streamer/capture
as local refs, and calls signal.throwIfAborted() after each await so an
interleaved stop() unwinds into a catch that tears down via the local refs and
clears instance state only if it still points at this attempt. isActive() now
reflects "starting" during the delay too.
Verified live: concurrent start is rejected ("이미 송출 중입니다"), stop() mid-
startup returns a cancel message with isActive=false and no uncaught error, and
the happy path still goes live and tears down cleanly. tsc --noEmit passes.
This commit is contained in:
@@ -37,10 +37,9 @@ export class SelfbotStreamer implements ScreenStreamer {
|
|||||||
* Wait a randomised, human-plausible amount of time. Resolves immediately if
|
* Wait a randomised, human-plausible amount of time. Resolves immediately if
|
||||||
* the stream is aborted (stop()) mid-wait, so teardown never hangs on a pause.
|
* the stream is aborted (stop()) mid-wait, so teardown never hangs on a pause.
|
||||||
*/
|
*/
|
||||||
private humanPause(minMs: number, maxMs: number): Promise<void> {
|
private humanPause(minMs: number, maxMs: number, signal?: AbortSignal): Promise<void> {
|
||||||
const ms = Math.floor(minMs + Math.random() * Math.max(0, maxMs - minMs));
|
const ms = Math.floor(minMs + Math.random() * Math.max(0, maxMs - minMs));
|
||||||
return new Promise((resolve) => {
|
return new Promise((resolve) => {
|
||||||
const signal = this.controller?.signal;
|
|
||||||
if (signal?.aborted) return resolve();
|
if (signal?.aborted) return resolve();
|
||||||
const onAbort = () => {
|
const onAbort = () => {
|
||||||
clearTimeout(timer);
|
clearTimeout(timer);
|
||||||
@@ -84,20 +83,38 @@ export class SelfbotStreamer implements ScreenStreamer {
|
|||||||
if (!ctx.voiceChannelId) {
|
if (!ctx.voiceChannelId) {
|
||||||
return "셀프봇 송출은 음성 채널 안에서 호출해야 합니다.";
|
return "셀프봇 송출은 음성 채널 안에서 호출해야 합니다.";
|
||||||
}
|
}
|
||||||
|
// Lock the starting state BEFORE any await: the human-pause delays below
|
||||||
|
// mean start() is in-flight for several seconds, so a second /stream call
|
||||||
|
// must be rejected by the `this.active` guard above, and the status must
|
||||||
|
// read "starting" rather than idle during the wait. Keep controller /
|
||||||
|
// streamer / capture as LOCAL refs so an interleaved stop() (which nulls the
|
||||||
|
// instance fields) can't turn our own continuation into a null dereference.
|
||||||
|
this.active = true;
|
||||||
|
const controller = (this.controller = new AbortController());
|
||||||
|
const signal = controller.signal;
|
||||||
|
let streamer: any = null;
|
||||||
|
let capture: ChildProcess | null = null;
|
||||||
|
|
||||||
|
try {
|
||||||
const { selfbot, vs } = await this.loadLib();
|
const { selfbot, vs } = await this.loadLib();
|
||||||
const { Streamer, prepareStream, playStream } = vs;
|
const { Streamer, prepareStream, playStream } = vs;
|
||||||
|
signal.throwIfAborted();
|
||||||
|
|
||||||
this.controller = new AbortController();
|
streamer = this.streamer = new Streamer(new selfbot.Client());
|
||||||
|
await streamer.client.login(this.config.selfbotToken);
|
||||||
this.streamer = new Streamer(new selfbot.Client());
|
signal.throwIfAborted();
|
||||||
await this.streamer.client.login(this.config.selfbotToken);
|
|
||||||
|
|
||||||
// Act like a person, not a bot: take a breath after coming online before
|
// Act like a person, not a bot: take a breath after coming online before
|
||||||
// navigating into the voice channel, then settle in for a few seconds
|
// navigating into the voice channel, then settle in for a few seconds
|
||||||
// before hitting "Go Live". Randomised so the cadence isn't fingerprintable.
|
// before hitting "Go Live". Randomised so the cadence isn't
|
||||||
await this.humanPause(900, 2200);
|
// fingerprintable. throwIfAborted() after each pause unwinds into the
|
||||||
await this.streamer.joinVoice(ctx.guildId, ctx.voiceChannelId);
|
// catch below if stop() lands mid-wait, so we never join/go-live on a
|
||||||
await this.humanPause(2500, 5000);
|
// torn-down streamer.
|
||||||
|
await this.humanPause(900, 2200, signal);
|
||||||
|
signal.throwIfAborted();
|
||||||
|
await streamer.joinVoice(ctx.guildId, ctx.voiceChannelId);
|
||||||
|
await this.humanPause(2500, 5000, signal);
|
||||||
|
signal.throwIfAborted();
|
||||||
|
|
||||||
const [w, h] = this.config.vncResolution.split("x").map((n) => parseInt(n, 10));
|
const [w, h] = this.config.vncResolution.split("x").map((n) => parseInt(n, 10));
|
||||||
|
|
||||||
@@ -118,7 +135,7 @@ export class SelfbotStreamer implements ScreenStreamer {
|
|||||||
const captureCodecArgs = hw
|
const captureCodecArgs = hw
|
||||||
? ["-c:v", "h264_nvenc", "-preset", "p4", "-tune", "ll", "-forced-idr", "1"]
|
? ["-c:v", "h264_nvenc", "-preset", "p4", "-tune", "ll", "-forced-idr", "1"]
|
||||||
: ["-c:v", "libx264", "-preset", "ultrafast", "-tune", "zerolatency"];
|
: ["-c:v", "libx264", "-preset", "ultrafast", "-tune", "zerolatency"];
|
||||||
const capture = spawn("ffmpeg", [
|
capture = this.capture = spawn("ffmpeg", [
|
||||||
"-loglevel", "error",
|
"-loglevel", "error",
|
||||||
"-f", "x11grab",
|
"-f", "x11grab",
|
||||||
"-framerate", String(this.config.vncFramerate),
|
"-framerate", String(this.config.vncFramerate),
|
||||||
@@ -131,9 +148,8 @@ export class SelfbotStreamer implements ScreenStreamer {
|
|||||||
"-g", String(this.config.vncFramerate),
|
"-g", String(this.config.vncFramerate),
|
||||||
"-f", "mpegts", "pipe:1",
|
"-f", "mpegts", "pipe:1",
|
||||||
]);
|
]);
|
||||||
this.capture = capture;
|
|
||||||
capture.stderr?.on("data", (d) => {
|
capture.stderr?.on("data", (d) => {
|
||||||
if (!this.controller?.signal.aborted) console.error("[selfbot x11grab]", d.toString().trim());
|
if (!signal.aborted) console.error("[selfbot x11grab]", d.toString().trim());
|
||||||
});
|
});
|
||||||
|
|
||||||
const { command, output } = prepareStream(
|
const { command, output } = prepareStream(
|
||||||
@@ -149,20 +165,45 @@ export class SelfbotStreamer implements ScreenStreamer {
|
|||||||
videoCodec: "H264",
|
videoCodec: "H264",
|
||||||
noTranscoding: true,
|
noTranscoding: true,
|
||||||
},
|
},
|
||||||
this.controller.signal,
|
signal,
|
||||||
);
|
);
|
||||||
command.on("error", (err: Error) => {
|
command.on("error", (err: Error) => {
|
||||||
if (!this.controller?.signal.aborted) console.error("[selfbot] ffmpeg error:", err);
|
if (!signal.aborted) console.error("[selfbot] ffmpeg error:", err);
|
||||||
});
|
});
|
||||||
|
signal.throwIfAborted();
|
||||||
|
|
||||||
this.active = true;
|
playStream(output, streamer, { type: "go-live" }, signal)
|
||||||
playStream(output, this.streamer, { type: "go-live" }, this.controller.signal)
|
.catch((err: Error) => {
|
||||||
.catch((err: Error) => console.error("[selfbot] playStream:", err))
|
if (!signal.aborted) console.error("[selfbot] playStream:", err);
|
||||||
|
})
|
||||||
.finally(() => {
|
.finally(() => {
|
||||||
this.active = false;
|
// The stream ended on its own (not via stop()); release the lock.
|
||||||
|
if (this.controller === controller) this.active = false;
|
||||||
});
|
});
|
||||||
|
|
||||||
return "🔴 셀프봇으로 VNC 화면을 음성채널에 실시간 송출 중입니다 (Go Live).";
|
return "🔴 셀프봇으로 VNC 화면을 음성채널에 실시간 송출 중입니다 (Go Live).";
|
||||||
|
} catch (e) {
|
||||||
|
// Startup was aborted (stop() during a pause) or failed. Tear down using
|
||||||
|
// our LOCAL refs, then clear instance state only if it still points at us
|
||||||
|
// (a concurrent stop()/start() may already have replaced it).
|
||||||
|
try {
|
||||||
|
capture?.kill("SIGKILL");
|
||||||
|
} catch {
|
||||||
|
/* ignore */
|
||||||
|
}
|
||||||
|
try {
|
||||||
|
streamer?.leaveVoice?.();
|
||||||
|
streamer?.client?.destroy?.();
|
||||||
|
} catch {
|
||||||
|
/* ignore */
|
||||||
|
}
|
||||||
|
if (this.capture === capture) this.capture = null;
|
||||||
|
if (this.streamer === streamer) this.streamer = null;
|
||||||
|
if (this.controller === controller) this.controller = null;
|
||||||
|
this.active = false;
|
||||||
|
if (signal.aborted) return "송출을 시작하는 중에 중지했습니다.";
|
||||||
|
throw e;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
async stop(): Promise<void> {
|
async stop(): Promise<void> {
|
||||||
|
|||||||
Reference in New Issue
Block a user