Processus de travail des files d'attente

Les processus de travail des files traitent les éléments des files de travail avec des onglets de navigateur concurrents. Chaque processus reçoit son propre onglet, dépile des éléments, exécute un script sur chacun, puis valide ou abandonne le résultat.

Concept

1. Vous mettez en file des éléments de travail : URL, enregistrements de données ou tâches.
2. Les processus dépilent les éléments et les traitent, chacun dans son propre onglet de navigateur.
3. Votre script par élément appelle g.worker.getContext() pour obtenir l’élément, fait son travail, puis appelle g.queue.commit() ou g.queue.abort().

Le panneau Workers affiche les pools actifs, les compteurs traités, les échecs, la profondeur de file et les contrôles pour mettre en pause ou arrêter le travail.

Deux façons de définir des processus

Déclarative : queues.json

Créez un fichier queues.json à la racine de votre espace de travail :

{
  "pools": [
    {
      "queue": "urls",
      "script": "scripts/process-url.js",
      "concurrency": 3,
      "throttle": 30,
      "layout": "horizontal",
      "clearTabs": true
    }
  ]
}

Champ	Obligatoire	Défaut	Description
queue	Oui	—	Nom de la file à traiter
script	Oui	—	Chemin du script relatif à la racine de l’espace de travail
concurrency	Non	1	Nombre de processus concurrents, donc d’onglets
throttle	Non	60	Nombre maximal de dépilages par minute et par processus
layout	Non	—	Disposition des onglets : horizontal, vertical ou quad
clearTabs	Non	false	Ferme tous les onglets existants avant de démarrer
requiresWorkflowEnvelope	Non	false	Exige que chaque élément de file porte les métadonnées du registre avant exécution

Démarrez un pool déclaré depuis la Queue Console, le panneau Queues compact ou les outils MCP.

Programmée : g.workers.start()

async function main() {
  // Enqueue some work
  for (let i = 1; i <= 100; i++) {
    g.queue.enqueue("pages", {
      url: "https://example.com/page/" + i
    });
  }

  // Start workers
  g.workers.start({
    clearTabs: true,
    layout: "horizontal",
    workers: [
      {
        queue: "pages",
        script: "scripts/scrape-page.js",
        concurrency: 4,
        throttle: { requestsPerMinute: 20 },
        requiresWorkflowEnvelope: false
      }
    ]
  });
}

g.workers.start() est fire-and-forget : les processus s’exécutent indépendamment du script qui les a lancés.

Script par élément

Chaque processus exécute votre script une fois par élément de file. Le script doit :

1. Appeler g.worker.getContext() pour obtenir l’élément et l’ID d’onglet.
2. Faire son travail.
3. Appeler g.queue.commit() ou g.queue.abort().

scripts/scrape-page.js

async function main() {
  const ctx = g.worker.getContext();
  const { item, tabId, queueName } = ctx;

  // Navigate in this worker's tab
  await g.nav.navigateToUrl(item.data.url, { browserTabId: tabId });
  await g.dom.waitForDomStable({ browserTabId: tabId });

  // Extract data
  const title = await g.dom.extractContent("h1", { browserTabId: tabId });

  // Store result
  g.store.put("results", item.data.url, {
    title: title,
    scrapedAt: new Date().toISOString()
  });

  // Mark complete
  g.queue.commit(item.id);
}

Il n’y a pas de validation automatique : vous devez valider ou abandonner chaque élément explicitement.

Gestion des échecs

Si votre script abandonne un élément et que celui-ci a dépassé maxRetries, l’élément passe dans la file des rejets (dead_{queueName}). Vous pouvez inspecter et réessayer les éléments rejetés :

const deadItems = g.queue.deadLetter("pages");
g.log("Failed items: " + deadItems.length);

// Retry a specific item
g.queue.retry("pages", deadItems[0].deadLetterId);

Contrôler les processus

Méthode	Description
g.workers.stop()	Arrête tous les processus
g.workers.stop("pages")	Arrête les processus d’une file précise
g.workers.pause()	Suspend le dépilage ; les éléments en cours se terminent
g.workers.resume()	Reprend le dépilage
g.workers.status()	Renvoie l’état de tous les pools

Le statut renvoie un tableau d’instantanés de pools :

{
  queue: string
  script: string
  concurrency: number
  activeWorkers: number
  processed: number
  failed: number
  remaining: number
  paused: boolean
}

Gestion des onglets

Les processus créent automatiquement un onglet de navigateur par processus concurrent. L’option layout les arrange :

• horizontal — onglets côte à côte
• vertical — onglets empilés
• quad — grille 2x2 pour quatre processus

Définissez clearTabs: true pour fermer tous les onglets existants avant le démarrage.

Limitation de débit

Le paramètre throttle contrôle la vitesse à laquelle chaque processus dépile les éléments. Une valeur de 30 signifie que chaque processus attend au moins 2 secondes entre deux dépilages. Avec concurrency: 3 et throttle: 30, le pool traite jusqu’à 90 éléments par minute au total.

Supervision et opérations

Chaque exécution d’élément par un processus apparaît comme une tâche dans le panneau Task Manager, avec durée en direct et boutons d’arrêt.

Pour les opérations quotidiennes sur les files, utilisez la Queue Console. Elle affiche les compteurs de files, éléments en attente, files des rejets, contenus sélectionnés, état des processus, contrôles run-one/start/stop et liens vers les éléments du Workflow Ledger lorsque les contenus de file portent des métadonnées de flux.

Si un script de processus échoue après avoir pris un élément, Guida abandonne la prise en charge quand c’est possible afin que l’élément de file ne reste pas bloqué. Arrêter une tâche Run One ou un pool doit arrêter le futur travail et ramener le travail de file en cours dans un état inspectable.

Lorsque des éléments de file sont liés à des éléments du registre de traçabilité, les chemins de réessai et run-one s’assurent d’abord que l’élément du registre peut être pris en charge. Si Guida ne peut pas réconcilier l’élément de file et l’état du registre, il affiche une erreur destinée à l’opérateur au lieu de créer du travail voué à l’échec.

Pour les files appartenant à un flux de travail, définissez requiresWorkflowEnvelope: true dans queues.json ou g.workers.start(). Guida refuse alors de traiter des éléments dont le contenu n’inclut pas une enveloppe de flux avec workflowName, runId, itemId et itemKey.

C’est utile pour les crawls longs où le travail en file et la vérité du Workflow Ledger doivent rester synchronisés. Le travail pris en charge mais invalide est bloqué et déplacé vers un état d’échec inspectable, au lieu de boucler sur des réessais sans issue.

Pour les grands pipelines qui couvrent plusieurs dossiers de flux, files, pools de processus et étapes du registre, modélisez l’intention de niveau opérateur dans Flux de travail. La Workflow Control Console peut alors montrer quelle file ou quel processus demande attention et ouvrir la Queue Console sur les données pertinentes.

Révision manuelle

Pour les files qui demandent un jugement humain, les vues en mode révision fournissent boutons d’action, raccourcis clavier et révision multi-onglet. Voir Files d’attente et révision.

Étapes suivantes

• Files d’attente et révision — gérer les files, inspecter les contenus et réviser les éléments
• Registre de traçabilité des flux de travail — inspecter l’état durable des flux et l’historique de récupération
• Flux de travail — décrire les opérations de niveau opérateur et les passages file/registre
• Mettre des URL en file et les traiter — exemple complet de mise en file, traitement et validation d’éléments
• Exécuter des processus parallèles — démarrer plusieurs processus navigateur depuis un script
• Référence API — g.queue.* — opérations sur les files
• Référence API — g.workers.* — méthodes de gestion des processus
• Espaces de travail — configuration de l’espace de travail pour queues.json