Queues in production not working

here's the main queue code , works locally

import { getOneRepoPackageJson } from "./getOneRepoLibraries.ts";
import { getGithubViewer } from "./getViewer.ts";
import { Edge } from "./getViewerRepos.ts";

interface EnqueueRepoPackagesCompoutemprops {
  repos: Edge[];
  viewer_token: string;
}

export async function enqueueRepoPackagesCompute(
  { repos, viewer_token }: EnqueueRepoPackagesCompoutemprops,
) {
  try {
    const kv = await Deno.openKv();
    const viewer = await getGithubViewer(viewer_token);

  console.log("-======= enqueueing repos for ====== ", viewer.name);
    const enqueueItems = async () => {
      for await (const [index, item] of repos.entries()) {
        const hundrecth = Math.floor(index / 50);
        // const delay = 1000 * 60 * (index + 1);

        if ("documentation_url" in item && "message" in item) return;
        const delay = Math.max(1, 100 * (index + 1) * (hundrecth + 1));
        console.log("=== QUEUE ITEM ==== ", item);
        console.log("=== DELAY === ", delay);
        await kv.enqueue({ message: "repo_pkgjson_queue", value: item }, {
          delay,
        });
      }
    };
    await enqueueItems();

   kv.listenQueue(async (msg) => {
      const data = msg as { message: string; value: Edge };

      if (!data.value) {
        return;
      }
      if (!data.message) {
        return;
      }
      if (data.message === "repo_pkgjson_queue") {
        console.log("writing data ==== ", data.value);
        const pkgjson = await getOneRepoPackageJson(
          data.value.node.nameWithOwner,
          viewer_token,
        );
        if (!pkgjson) return;
        if ("documentation_url" in pkgjson || "message" in pkgjson) return;
        await kv.set([
          "repo_pkgjson",
          viewer.name,
          data.value.node.nameWithOwner,
        ], pkgjson);
      }
      return;
    });
  } catch (error) {
    throw error;
  }
}

import { getOneRepoPackageJson } from "./getOneRepoLibraries.ts";
import { getGithubViewer } from "./getViewer.ts";
import { Edge } from "./getViewerRepos.ts";

interface EnqueueRepoPackagesCompoutemprops {
  repos: Edge[];
  viewer_token: string;
}

export async function enqueueRepoPackagesCompute(
  { repos, viewer_token }: EnqueueRepoPackagesCompoutemprops,
) {
  try {
    const kv = await Deno.openKv();
    const viewer = await getGithubViewer(viewer_token);

  console.log("-======= enqueueing repos for ====== ", viewer.name);
    const enqueueItems = async () => {
      for await (const [index, item] of repos.entries()) {
        const hundrecth = Math.floor(index / 50);
        // const delay = 1000 * 60 * (index + 1);

        if ("documentation_url" in item && "message" in item) return;
        const delay = Math.max(1, 100 * (index + 1) * (hundrecth + 1));
        console.log("=== QUEUE ITEM ==== ", item);
        console.log("=== DELAY === ", delay);
        await kv.enqueue({ message: "repo_pkgjson_queue", value: item }, {
          delay,
        });
      }
    };
    await enqueueItems();

   kv.listenQueue(async (msg) => {
      const data = msg as { message: string; value: Edge };

      if (!data.value) {
        return;
      }
      if (!data.message) {
        return;
      }
      if (data.message === "repo_pkgjson_queue") {
        console.log("writing data ==== ", data.value);
        const pkgjson = await getOneRepoPackageJson(
          data.value.node.nameWithOwner,
          viewer_token,
        );
        if (!pkgjson) return;
        if ("documentation_url" in pkgjson || "message" in pkgjson) return;
        await kv.set([
          "repo_pkgjson",
          viewer.name,
          data.value.node.nameWithOwner,
        ], pkgjson);
      }
      return;
    });
  } catch (error) {
    throw error;
  }
}

the rest of the code it is at https://github.com/tigawanna/library-stats

GitHub

GitHub - tigawanna/library-stats

Contribute to tigawanna/library-stats development by creating an account on GitHub.

2 Replies

Deno AI Helper•8mo ago

It seems you're encountering issues with your queue code when deploying it to production, even though it works locally. Let's go through some steps to troubleshoot and ensure that your code aligns with the best practices outlined in the documentation. 1. Check for --unstable flag: According to the documentation, Deno KV is currently experimental and requires the --unstable flag when executing Deno programs that use KV. Make sure that your production environment includes this flag when running your Deno application. 2. Verify Deno version: Ensure that the Deno version in production matches the one you're using locally. Since Deno KV is in beta, there might be changes between versions that could affect the behavior of your queues. 3. Review Queue Behavior: The documentation states that the runtime guarantees at-least-once delivery for enqueued messages. This means that your listenQueue handler should be designed to handle duplicate messages correctly. Make sure that your production code is idempotent and can handle such cases without causing issues. 4. Error Handling: Your listenQueue handler should be robust against errors. If an exception is thrown, the runtime will automatically retry the handler until it succeeds or reaches the maximum retry attempts. Ensure that your error handling in production is adequate to prevent infinite retry loops or message loss. 5. KV Atomic Transactions: If your application logic requires it, consider using KV atomic transactions to ensure that updates on each message are performed only once. This can help prevent race conditions and ensure data consistency. 6. Environment Variables: If your production environment uses different configurations or environment variables, double-check that all necessary

Deno AI Helper•8mo ago

variables (like viewer_token) are correctly set and accessible. 7. Logging and Monitoring: Increase the logging verbosity in your production environment to capture more details about the queue's behavior. This can help you identify where the process might be failing. 8. Network and Permissions: Ensure that your production environment has the necessary network access and permissions to interact with any external services or APIs that your queue handlers might be calling. 9. Deno Deploy: If you're using Deno Deploy, remember that it optimizes queue processing by spinning up V8 isolates on-demand. Make sure that your code is compatible with this environment and that you're not relying on any state that might not persist across isolate invocations. If after checking all these points your queues are still not working in production, you might want to isolate the issue by simplifying your queue processing logic and testing it incrementally in the production environment. This can help pinpoint the exact step where the process breaks down. Remember to consult the official Deno documentation for the most up-to-date information and best practices regarding Deno KV and queues. 🦕

Helpful Links: Offload webhook processing to a queue | Deno Docs Using Queues | Deno Docs Announcing Deno Queues Struggling with interactive FreshCharts - Deno