bot/mod.ts

@@ -26,27 +26,6 @@ bot.use(session);
 
 bot.api.config.use(GrammyFiles.hydrateFiles(bot.token));
 
-// Automatically cancel requests after 30 seconds
-bot.api.config.use(async (prev, method, payload, signal) => {
-  const controller = new AbortController();
-  let timedOut = false;
-  const timeout = setTimeout(() => {
-    timedOut = true;
-    controller.abort();
-  }, 30 * 1000);
-  signal?.addEventListener("abort", () => {
-    controller.abort();
-  });
-  try {
-    return await prev(method, payload, controller.signal);
-  } finally {
-    clearTimeout(timeout);
-    if (timedOut) {
-      logger().warning(`${method} timed out`);
-    }
-  }
-});
-
 // Automatically retry bot requests if we get a "too many requests" or telegram internal error
 bot.api.config.use(async (prev, method, payload, signal) => {
   const maxAttempts = payload && ("maxAttempts" in payload) ? payload.maxAttempts ?? 3 : 3;

bot/queueCommand.ts

@@ -39,8 +39,8 @@ export async function queueCommand(ctx: Grammy.CommandContext<Context>) {
         : ["Queue is empty.\n"],
       "\nActive workers:\n",
       ...ctx.session.global.workers.flatMap((worker) => [
-        runningWorkers.has(worker.id) ? "✅ " : "☠️ ",
-        fmt`${bold(worker.name || worker.id)} `,
+        runningWorkers.has(worker.name) ? "✅ " : "☠️ ",
+        fmt`${bold(worker.name)} `,
         `(max ${(worker.maxResolution / 1000000).toFixed(1)} Mpx) `,
         "\n",
       ]),
@@ -49,7 +49,7 @@ export async function queueCommand(ctx: Grammy.CommandContext<Context>) {
 
   async function handleFutureUpdates() {
     for (let idx = 0; idx < 30; idx++) {
-      await ctx.api.sendChatAction(ctx.chat.id, "typing", { maxAttempts: 1 } as never);
+      await ctx.api.sendChatAction(ctx.chat.id, "typing");
       await new Promise((resolve) => setTimeout(resolve, 4000));
       const nextFormattedMessage = await getMessageText();
       if (nextFormattedMessage.text !== formattedMessage.text) {

bot/session.ts

@@ -20,9 +20,9 @@ export interface GlobalData {
 }
 
 export interface WorkerData {
-  id: string;
-  name?: string;
+  name: string;
   api: SdApi;
+  auth?: string;
   maxResolution: number;
 }
 
@@ -52,7 +52,7 @@ const getDefaultGlobalData = (): GlobalData => ({
   },
   workers: [
     {
-      id: "local",
+      name: "local",
       api: { url: Deno.env.get("SD_API_URL") ?? "http://127.0.0.1:7860/" },
       maxResolution: 1024 * 1024,
     },

sd.ts

@@ -1,17 +1,13 @@
 import { Async, AsyncX, PngChunksExtract, PngChunkText } from "./deps.ts";
 
+const neverSignal = new AbortController().signal;
+
 export interface SdApi {
   url: string;
   auth?: string;
 }
 
-async function fetchSdApi<T>(
-  api: SdApi,
-  endpoint: string,
-  { body, timeoutMs }: { body?: unknown; timeoutMs?: number } = {},
-): Promise<T> {
-  const controller = new AbortController();
-  const timeoutId = timeoutMs ? setTimeout(() => controller.abort(), timeoutMs) : undefined;
+async function fetchSdApi<T>(api: SdApi, endpoint: string, body?: unknown): Promise<T> {
   let options: RequestInit | undefined;
   if (body != null) {
     options = {
@@ -21,18 +17,13 @@ async function fetchSdApi<T>(
         ...api.auth ? { Authorization: api.auth } : {},
       },
       body: JSON.stringify(body),
-      signal: controller.signal,
     };
   } else if (api.auth) {
     options = {
       headers: { Authorization: api.auth },
-      signal: controller.signal,
     };
   }
   const response = await fetch(new URL(endpoint, api.url), options).catch(() => {
-    if (controller.signal.aborted) {
-      throw new SdApiError(endpoint, options, -1, "Timed out");
-    }
     throw new SdApiError(endpoint, options, 0, "Network error");
   });
   const result = await response.json().catch(() => {
@@ -40,7 +31,6 @@ async function fetchSdApi<T>(
       detail: "Invalid JSON",
     });
   });
-  clearTimeout(timeoutId);
   if (!response.ok) {
     throw new SdApiError(endpoint, options, response.status, response.statusText, result);
   }
@@ -88,12 +78,9 @@ export async function sdTxt2Img(
   api: SdApi,
   params: Partial<SdTxt2ImgRequest>,
   onProgress?: (progress: SdProgressResponse) => void,
+  signal: AbortSignal = neverSignal,
 ): Promise<SdResponse<SdTxt2ImgRequest>> {
-  const request = fetchSdApi<SdResponse<SdTxt2ImgRequest>>(
-    api,
-    "sdapi/v1/txt2img",
-    { body: params },
-  )
+  const request = fetchSdApi<SdResponse<SdTxt2ImgRequest>>(api, "sdapi/v1/txt2img", params)
     // JSON field "info" is a JSON-serialized string so we need to parse this part second time
     .then((data) => ({
       ...data,
@@ -102,15 +89,13 @@ export async function sdTxt2Img(
 
   try {
     while (true) {
-      await Promise.race([request, Async.delay(3000)]);
+      await Async.abortable(Promise.race([request, Async.delay(4000)]), signal);
       if (await AsyncX.promiseState(request) !== "pending") return await request;
-      onProgress?.(
-        await fetchSdApi<SdProgressResponse>(api, "sdapi/v1/progress", { timeoutMs: 10_000 }),
-      );
+      onProgress?.(await fetchSdApi<SdProgressResponse>(api, "sdapi/v1/progress"));
     }
   } finally {
     if (await AsyncX.promiseState(request) === "pending") {
-      await fetchSdApi(api, "sdapi/v1/interrupt", { timeoutMs: 10_000 });
+      await fetchSdApi(api, "sdapi/v1/interrupt", {});
     }
   }
 }
@@ -133,12 +118,9 @@ export async function sdImg2Img(
   api: SdApi,
   params: Partial<SdImg2ImgRequest>,
   onProgress?: (progress: SdProgressResponse) => void,
+  signal: AbortSignal = neverSignal,
 ): Promise<SdResponse<SdImg2ImgRequest>> {
-  const request = fetchSdApi<SdResponse<SdImg2ImgRequest>>(
-    api,
-    "sdapi/v1/img2img",
-    { body: params },
-  )
+  const request = fetchSdApi<SdResponse<SdImg2ImgRequest>>(api, "sdapi/v1/img2img", params)
     // JSON field "info" is a JSON-serialized string so we need to parse this part second time
     .then((data) => ({
       ...data,
@@ -147,15 +129,13 @@ export async function sdImg2Img(
 
   try {
     while (true) {
-      await Promise.race([request, Async.delay(3000)]);
+      await Async.abortable(Promise.race([request, Async.delay(4000)]), signal);
       if (await AsyncX.promiseState(request) !== "pending") return await request;
-      onProgress?.(
-        await fetchSdApi<SdProgressResponse>(api, "sdapi/v1/progress", { timeoutMs: 10_000 }),
-      );
+      onProgress?.(await fetchSdApi<SdProgressResponse>(api, "sdapi/v1/progress"));
     }
   } finally {
     if (await AsyncX.promiseState(request) === "pending") {
-      await fetchSdApi(api, "sdapi/v1/interrupt", { timeoutMs: 10_000 });
+      await fetchSdApi(api, "sdapi/v1/interrupt", {});
     }
   }
 }
@@ -240,7 +220,7 @@ export interface SdProgressState {
 }
 
 export function sdGetConfig(api: SdApi): Promise<SdConfigResponse> {
-  return fetchSdApi(api, "config", { timeoutMs: 10_000 });
+  return fetchSdApi(api, "config");
 }
 
 export interface SdConfigResponse {

tasks/pingWorkers.ts

@@ -15,13 +15,13 @@ export async function pingWorkers(): Promise<never> {
       const config = await getGlobalSession();
       for (const worker of config.workers) {
         const status = await sdGetConfig(worker.api).catch(() => null);
-        const wasRunning = runningWorkers.has(worker.id);
+        const wasRunning = runningWorkers.has(worker.name);
         if (status) {
-          runningWorkers.add(worker.id);
-          if (!wasRunning) logger().info(`Worker ${worker.id} is online`);
+          runningWorkers.add(worker.name);
+          if (!wasRunning) logger().info(`Worker ${worker.name} is online`);
         } else {
-          runningWorkers.delete(worker.id);
-          if (wasRunning) logger().warning(`Worker ${worker.id} went offline`);
+          runningWorkers.delete(worker.name);
+          if (wasRunning) logger().warning(`Worker ${worker.name} went offline`);
         }
       }
       await Async.delay(60 * 1000);

tasks/processJobs.ts

@@ -34,8 +34,8 @@ export async function processJobs(): Promise<never> {
       // find a worker to handle the job
       const config = await getGlobalSession();
       const worker = config.workers?.find((worker) =>
-        runningWorkers.has(worker.id) &&
-        !busyWorkers.has(worker.id)
+        runningWorkers.has(worker.name) &&
+        !busyWorkers.has(worker.name)
       );
       if (!worker) continue;
 
@@ -45,16 +45,16 @@ export async function processJobs(): Promise<never> {
         status: {
           type: "processing",
           progress: 0,
-          worker: worker.id,
+          worker: worker.name,
           updatedDate: new Date(),
           message: job.value.status.type !== "done" ? job.value.status.message : undefined,
         },
       }));
-      busyWorkers.add(worker.id);
+      busyWorkers.add(worker.name);
       processJob(job, worker, config)
         .catch(async (err) => {
           logger().error(
-            `Job failed for ${formatUserChat(job.value)} via ${worker.id}: ${err}`,
+            `Job failed for ${formatUserChat(job.value)} via ${worker.name}: ${err}`,
           );
           if (err instanceof Grammy.GrammyError || err instanceof SdApiError) {
             await bot.api.sendMessage(
@@ -72,9 +72,9 @@ export async function processJobs(): Promise<never> {
               err.statusCode === 401
             )
           ) {
-            runningWorkers.delete(worker.id);
+            runningWorkers.delete(worker.name);
             logger().warning(
-              `Worker ${worker.id} was marked as offline because of network error`,
+              `Worker ${worker.name} was marked as offline because of network error`,
             );
           }
           await job.delete().catch(() => undefined);
@@ -82,7 +82,7 @@ export async function processJobs(): Promise<never> {
             await jobStore.create(job.value);
           }
         })
-        .finally(() => busyWorkers.delete(worker.id));
+        .finally(() => busyWorkers.delete(worker.name));
     } catch (err) {
       logger().warning(`Processing jobs failed: ${err}`);
     }
@@ -91,7 +91,7 @@ export async function processJobs(): Promise<never> {
 
 async function processJob(job: IKV.Model<JobSchema>, worker: WorkerData, config: GlobalData) {
   logger().debug(
-    `Job started for ${formatUserChat(job.value)} using ${worker.id}`,
+    `Job started for ${formatUserChat(job.value)} using ${worker.name}`,
   );
   const startDate = new Date();
 
@@ -161,9 +161,7 @@ async function processJob(job: IKV.Model<JobSchema>, worker: WorkerData, config:
 
   // process the job
   const handleProgress = async (progress: SdProgressResponse) => {
-    // Important: don't let any errors escape this function
     if (job.value.status.type === "processing" && job.value.status.message) {
-      if (job.value.status.progress === progress.progress) return;
       await Promise.all([
         bot.api.sendChatAction(job.value.chat.id, "upload_photo", { maxAttempts: 1 }),
         bot.api.editMessageText(
@@ -179,16 +177,12 @@ async function processJob(job: IKV.Model<JobSchema>, worker: WorkerData, config:
           status: {
             type: "processing",
             progress: progress.progress,
-            worker: worker.id,
+            worker: worker.name,
             updatedDate: new Date(),
             message: value.status.type !== "done" ? value.status.message : undefined,
           },
         }), { maxAttempts: 1 }),
-      ]).catch((err) =>
-        logger().warning(
-          `Updating job status for ${formatUserChat(job.value)} using ${worker.id} failed: ${err}`,
-        )
-      );
+      ]).catch(() => undefined);
     }
   };
   let response: SdResponse<unknown>;
@@ -223,7 +217,6 @@ async function processJob(job: IKV.Model<JobSchema>, worker: WorkerData, config:
       job.value.status.message.chat.id,
       job.value.status.message.message_id,
       `Uploading your images...`,
-      { maxAttempts: 1 },
     ).catch(() => undefined);
   }
 
@@ -244,7 +237,7 @@ async function processJob(job: IKV.Model<JobSchema>, worker: WorkerData, config:
         fmt`${bold("CFG scale:")} ${response.info.cfg_scale}, `,
         fmt`${bold("Seed:")} ${response.info.seed}, `,
         fmt`${bold("Size")}: ${response.info.width}x${response.info.height}, `,
-        fmt`${bold("Worker")}: ${worker.id}, `,
+        fmt`${bold("Worker")}: ${worker.name}, `,
         fmt`${bold("Time taken")}: ${FmtDuration.format(jobDurationMs, { ignoreZero: true })}`,
       ]
       : [],
@@ -282,11 +275,7 @@ async function processJob(job: IKV.Model<JobSchema>, worker: WorkerData, config:
       });
       break;
     } catch (err) {
-      logger().warning(
-        `Sending images (attempt ${sendMediaAttempt}) for ${
-          formatUserChat(job.value)
-        } using ${worker.id} failed: ${err}`,
-      );
+      logger().warning(`Sending images (attempt ${sendMediaAttempt}) failed: ${err}`);
       if (sendMediaAttempt >= 6) throw err;
       // wait 2 * 5 seconds before retrying
       for (let i = 0; i < 2; i++) {
@@ -324,7 +313,7 @@ async function processJob(job: IKV.Model<JobSchema>, worker: WorkerData, config:
   }));
 
   logger().debug(
-    `Job finished for ${formatUserChat(job.value)} using ${worker.id}${
+    `Job finished for ${formatUserChat(job.value)} using ${worker.name}${
       sendMediaAttempt > 1 ? ` after ${sendMediaAttempt} attempts` : ""
     }`,
   );