Cancel searches when clients leave

application_neo4j/README.md

@@ -145,3 +145,11 @@ La création initiale des projections GDS reste sérialisée afin que deux
 requêtes arrivant au démarrage ne tentent pas de créer le même graphe. Une fois
 les projections disponibles, les BFS et la construction de leurs résultats
 s'exécutent indépendamment.
+
+Lorsqu'un utilisateur quitte la page pendant une recherche, le navigateur
+envoie une demande d'annulation. Un job encore en file est retiré
+immédiatement. Pour un job actif, l'application termine la transaction Neo4j
+identifiée par les métadonnées du job, puis vérifie aussi un drapeau
+d'annulation entre les lots de construction du résultat. Le signal envoyé par
+le navigateur est une garantie au mieux : une fermeture brutale du processus
+ou une coupure réseau peut empêcher son émission.

application_neo4j/app.py

@@ -1,7 +1,7 @@
 # --- Import et initialisation Flask ---
 import os
 from flask import Flask, request, render_template, jsonify, session, redirect, url_for
-from neo4j import GraphDatabase, basic_auth
+from neo4j import GraphDatabase, Query, basic_auth
 from graphdatascience import GraphDataScience
 import app_algorithms as algo
 from translations import t
@@ -51,6 +51,21 @@ search_jobs = {}
 search_jobs_lock = Lock()
 graph_projection_lock = Lock()
 
+
+class SearchCancelled(Exception):
+    """Raised by a worker when its browser no longer needs the result."""
+
+
+def search_cancel_requested(job_id):
+    with search_jobs_lock:
+        job = search_jobs.get(job_id)
+        return bool(job and job.get("cancel_requested"))
+
+
+def raise_if_search_cancelled(job_id):
+    if search_cancel_requested(job_id):
+        raise SearchCancelled()
+
 # --- Configuration des arguments du script ---
 parser = argparse.ArgumentParser(description="Script pour lancer la web application.")
 
@@ -212,6 +227,7 @@ def inject_i18n():
         "search.progress_stage_relationships", "search.progress_stage_formatting",
         "search.progress_stage_highlights",
         "search.progress_stage_completed", "search.progress_stage_failed",
+        "search.progress_stage_cancelled",
         "search.progress_elapsed", "search.progress_items",
         "search.progress_server_percent",
         "search.progress_remaining_step", "search.progress_server_note",
@@ -343,6 +359,7 @@ def execute_search_job(job_id, name, depth, is_unlimited, filters, expert, lang)
     graph_data = result["graph_data"]
 
     try:
+        raise_if_search_cancelled(job_id)
         update_search_job(job_id, status="running", started_at=time())
         def report_gds_stage(stage, gds_job_id):
             set_search_stage(job_id, stage, gds_job_id)
@@ -354,6 +371,7 @@ def execute_search_job(job_id, name, depth, is_unlimited, filters, expert, lang)
             search_job_id=job_id,
             progress_callback=report_gds_stage,
         )
+        raise_if_search_cancelled(job_id)
 
         natural_graph_name = f"{GDS_GRAPH_NAME}_natural"
         reverse_graph_name = f"{GDS_GRAPH_NAME}_reverse"
@@ -367,6 +385,8 @@ def execute_search_job(job_id, name, depth, is_unlimited, filters, expert, lang)
             expert,
             job_id=job_id,
             progress_callback=report_gds_stage,
+            neo4j_driver=driver,
+            cancel_check=lambda: raise_if_search_cancelled(job_id),
         )
 
         if not gds_result:
@@ -385,6 +405,8 @@ def execute_search_job(job_id, name, depth, is_unlimited, filters, expert, lang)
             process_gds_bfs_results(
                 gds_result,
                 graph_data,
+                job_id=job_id,
+                cancel_check=lambda: raise_if_search_cancelled(job_id),
                 progress_callback=report_result_progress,
             )
 
@@ -392,6 +414,7 @@ def execute_search_job(job_id, name, depth, is_unlimited, filters, expert, lang)
                 if not graph_data["nodes"] and not graph_data["edges"]:
                     result["message"] = t("error.no_neighbors", lang, name=name)
             elif gds_result["source_label"] == "Model":
+                raise_if_search_cancelled(job_id)
                 set_search_stage(job_id, "building_highlights")
                 result["highlights"] = algo.get_genealogy_highlights(
                     gds, name, lang=lang
@@ -401,6 +424,7 @@ def execute_search_job(job_id, name, depth, is_unlimited, filters, expert, lang)
             elif not graph_data["nodes"] and not graph_data["edges"]:
                 result["message"] = t("error.no_neighbors", lang, name=name)
 
+        raise_if_search_cancelled(job_id)
         update_search_job(
             job_id,
             status="completed",
@@ -412,7 +436,32 @@ def execute_search_job(job_id, name, depth, is_unlimited, filters, expert, lang)
             completed_at=time(),
             result=result,
         )
+    except SearchCancelled:
+        update_search_job(
+            job_id,
+            status="cancelled",
+            stage="cancelled",
+            gds_job_id=None,
+            application_progress_percent=None,
+            completed_items=None,
+            total_items=None,
+            completed_at=time(),
+            result=None,
+        )
     except Exception as error:
+        if search_cancel_requested(job_id):
+            update_search_job(
+                job_id,
+                status="cancelled",
+                stage="cancelled",
+                gds_job_id=None,
+                application_progress_percent=None,
+                completed_items=None,
+                total_items=None,
+                completed_at=time(),
+                result=None,
+            )
+            return
         if "Failed to find a node" in str(error):
             message = t("error.node_not_found", lang, name=name)
         else:
@@ -460,6 +509,38 @@ def get_gds_progress(gds_job_id):
         return None
 
 
+def terminate_search_transactions(job_id):
+    """Terminate any active Neo4j transaction tagged for this search."""
+    try:
+        with driver.session() as neo4j_session:
+            records = neo4j_session.run(
+                """
+                SHOW TRANSACTIONS
+                YIELD transactionId, metaData, status
+                WHERE metaData.search_job_id = $job_id
+                  AND NOT status STARTS WITH 'Terminated'
+                RETURN transactionId
+                """,
+                {"job_id": job_id},
+            )
+            transaction_ids = [
+                record["transactionId"] for record in records
+            ]
+            if transaction_ids:
+                neo4j_session.run(
+                    """
+                    TERMINATE TRANSACTIONS $transaction_ids
+                    YIELD transactionId, message
+                    RETURN transactionId, message
+                    """,
+                    {"transaction_ids": transaction_ids},
+                ).consume()
+    except Exception as error:
+        # The cooperative cancellation flag still stops the worker between
+        # batches if transaction termination is unavailable.
+        print(f"Could not terminate transactions for {job_id}: {error}")
+
+
 @app.route("/api/search-jobs", methods=["POST"])
 def create_search_job():
     cleanup_search_jobs()
@@ -500,9 +581,11 @@ def create_search_job():
             "total_items": None,
             "lang": lang,
             "result": None,
+            "cancel_requested": False,
+            "future": None,
         }
 
-    search_executor.submit(
+    future = search_executor.submit(
         execute_search_job,
         job_id,
         name,
@@ -512,11 +595,55 @@ def create_search_job():
         expert,
         lang,
     )
+    with search_jobs_lock:
+        job = search_jobs.get(job_id)
+        if job:
+            job["future"] = future
+            cancel_requested = job.get("cancel_requested")
+        else:
+            cancel_requested = True
+    if cancel_requested and future.cancel():
+        update_search_job(
+            job_id,
+            status="cancelled",
+            stage="cancelled",
+            completed_at=time(),
+        )
 
     return jsonify({
         "job_id": job_id,
         "status_url": url_for("search_job_status", job_id=job_id),
         "result_url": url_for("search_job_result", job_id=job_id, lang=lang),
+        "cancel_url": url_for("cancel_search_job", job_id=job_id),
+    }), 202
+
+
+@app.route("/api/search-jobs/<job_id>/cancel", methods=["POST"])
+def cancel_search_job(job_id):
+    with search_jobs_lock:
+        job = search_jobs.get(job_id)
+        if not job:
+            return jsonify({"error": "Search job not found"}), 404
+        if job["status"] in ("completed", "failed", "cancelled"):
+            return jsonify({"status": job["status"]})
+        job["cancel_requested"] = True
+        job["cancel_requested_at"] = time()
+        future = job.get("future")
+
+    cancelled_before_start = bool(future and future.cancel())
+    if cancelled_before_start:
+        update_search_job(
+            job_id,
+            status="cancelled",
+            stage="cancelled",
+            completed_at=time(),
+            result=None,
+        )
+    else:
+        terminate_search_transactions(job_id)
+
+    return jsonify({
+        "status": "cancelled" if cancelled_before_start else "cancelling"
     }), 202
 
 
@@ -747,6 +874,8 @@ def findnode_expert():
 def process_gds_bfs_results(
     gds_result: Dict,
     graph_data: Dict,
+    job_id=None,
+    cancel_check=None,
     progress_callback=None,
 ):
     """
@@ -805,6 +934,8 @@ def process_gds_bfs_results(
 
     with driver.session() as session:
         for batch_start in range(0, total_nodes, RESULT_BATCH_SIZE):
+            if cancel_check:
+                cancel_check()
             batch_ids = discovered_node_ids[
                 batch_start:batch_start + RESULT_BATCH_SIZE
             ]
@@ -812,7 +943,7 @@ def process_gds_bfs_results(
                 {"id": node_id, "distance": distance_by_node[node_id]}
                 for node_id in batch_ids
             ]
-            record = session.run(
+            query = Query(
                 """
                 UNWIND $nodes AS node_info
                 MATCH (n) WHERE id(n) = node_info.id
@@ -848,6 +979,10 @@ def process_gds_bfs_results(
                     distance: node_info.distance
                 }) AS nodes_data
                 """,
+                metadata={"search_job_id": job_id} if job_id else None,
+            )
+            record = session.run(
+                query,
                 {"nodes": batch_nodes},
             ).single()
             if record:
@@ -869,7 +1004,9 @@ def process_gds_bfs_results(
         if progress_callback:
             progress_callback("building_relationships", 0, 0)
 
-        count_record = session.run(
+        if cancel_check:
+            cancel_check()
+        count_query = Query(
             """
             UNWIND $all_ids AS source_id
             MATCH (source) WHERE id(source) = source_id
@@ -877,6 +1014,10 @@ def process_gds_bfs_results(
             WHERE id(target) IN $all_ids
             RETURN count(relationship) AS relationship_count
             """,
+            metadata={"search_job_id": job_id} if job_id else None,
+        )
+        count_record = session.run(
+            count_query,
             {"all_ids": discovered_node_ids},
         ).single()
         total_relationships = (
@@ -889,10 +1030,12 @@ def process_gds_bfs_results(
             )
 
         for batch_start in range(0, total_nodes, RESULT_BATCH_SIZE):
+            if cancel_check:
+                cancel_check()
             source_ids = discovered_node_ids[
                 batch_start:batch_start + RESULT_BATCH_SIZE
             ]
-            record = session.run(
+            relationship_query = Query(
                 """
                 UNWIND $source_ids AS source_id
                 MATCH (source) WHERE id(source) = source_id
@@ -904,6 +1047,10 @@ def process_gds_bfs_results(
                     targetName: endNode(relationship).name
                 }) AS relationships
                 """,
+                metadata={"search_job_id": job_id} if job_id else None,
+            )
+            record = session.run(
+                relationship_query,
                 {
                     "source_ids": source_ids,
                     "all_ids": discovered_node_ids,
@@ -930,6 +1077,8 @@ def process_gds_bfs_results(
                 total_relationships,
             )
 
+    if cancel_check:
+        cancel_check()
     if progress_callback:
         progress_callback("formatting_result", 0, 0)
 

application_neo4j/app_algorithms.py

@@ -3,6 +3,7 @@ from graphdatascience import GraphDataScience
 from typing import Dict, List, Any
 import pandas as pd
 from translations import t
+from neo4j import Query
 
 def run_gds_bfs(
     gds: GraphDataScience,
@@ -13,6 +14,8 @@ def run_gds_bfs(
     expert=False,
     job_id: str = None,
     progress_callback=None,
+    neo4j_driver=None,
+    cancel_check=None,
 ) -> Dict[str, Any]:
     """
     Parcourt les descendants et ascendants avec le fork OpenGDS.
@@ -53,17 +56,45 @@ def run_gds_bfs(
     if max_depth is not None:
         bfs_params["maxDepth"] = max_depth
 
-    g_natural = gds.graph.get(natural_graph_name)
+    def run_bfs(graph_name, params):
+        if cancel_check:
+            cancel_check()
+        if neo4j_driver is None:
+            graph = gds.graph.get(graph_name)
+            result = gds.bfs.stream(graph, **params)
+        else:
+            query = Query(
+                """
+                CALL gds.bfs.stream($graph_name, $configuration)
+                YIELD nodeIds, depths
+                RETURN nodeIds, depths
+                """,
+                metadata={"search_job_id": job_id},
+            )
+            with neo4j_driver.session() as neo4j_session:
+                records = neo4j_session.run(
+                    query,
+                    {
+                        "graph_name": graph_name,
+                        "configuration": params,
+                    },
+                )
+                result = pd.DataFrame(
+                    record.data() for record in records
+                )
+        if cancel_check:
+            cancel_check()
+        return result
+
     desc_job_id = f"{job_id}-descendants" if job_id else None
     if desc_job_id:
         bfs_params["jobId"] = desc_job_id
     if progress_callback:
         progress_callback("searching_descendants", desc_job_id)
-    desc_df = gds.bfs.stream(g_natural, **bfs_params)
+    desc_df = run_bfs(natural_graph_name, bfs_params)
     _validate_depths_result(desc_df)
     print("BFS descendants terminé avec les profondeurs.")
 
-    g_reverse = gds.graph.get(reverse_graph_name)
     asc_job_id = f"{job_id}-ancestors" if job_id else None
     if asc_job_id:
         bfs_params["jobId"] = asc_job_id
@@ -71,7 +102,7 @@ def run_gds_bfs(
         del bfs_params["jobId"]
     if progress_callback:
         progress_callback("searching_ancestors", asc_job_id)
-    asc_df = gds.bfs.stream(g_reverse, **bfs_params)
+    asc_df = run_bfs(reverse_graph_name, bfs_params)
     _validate_depths_result(asc_df)
     print("BFS ascendants terminé avec les profondeurs.")
 

application_neo4j/static/js/search_progress.js

@@ -1,5 +1,6 @@
 (() => {
   const POLL_INTERVAL_MS = 750;
+  let activeCancelUrl = null;
 
   function translate(key, fallback) {
     return window.__I18N_DATA?.[key] || fallback;
@@ -25,6 +26,7 @@
       building_highlights: ["search.progress_stage_highlights", "Calcul des modèles importants…"],
       completed: ["search.progress_stage_completed", "Recherche terminée."],
       failed: ["search.progress_stage_failed", "La recherche a échoué."],
+      cancelled: ["search.progress_stage_cancelled", "Recherche annulée."],
     };
     const [key, fallback] = labels[stage] || labels.queued;
     return translate(key, fallback);
@@ -58,6 +60,40 @@
     setSubmitting(form, false);
   }
 
+  function showCancelled(form, progress) {
+    const stage = progress.querySelector(".search-progress-stage");
+    const status = progress.querySelector(".search-progress-status");
+    const note = progress.querySelector(".search-progress-note");
+    const bar = progress.querySelector(".search-progress-bar");
+    stage.textContent = stageLabel("cancelled");
+    status.textContent = "";
+    note.textContent = "";
+    bar.style.width = "0%";
+    progress.classList.remove("alert-info", "alert-danger");
+    progress.classList.add("alert-secondary");
+    setSubmitting(form, false);
+  }
+
+  function cancelActiveSearch() {
+    if (!activeCancelUrl) return;
+    const cancelUrl = activeCancelUrl;
+    activeCancelUrl = null;
+    if (
+      navigator.sendBeacon
+      && navigator.sendBeacon(
+        cancelUrl,
+        new Blob([], { type: "text/plain" })
+      )
+    ) {
+      return;
+    }
+    fetch(cancelUrl, {
+      method: "POST",
+      credentials: "same-origin",
+      keepalive: true,
+    }).catch(() => {});
+  }
+
   function renderStatus(progress, job) {
     const stage = progress.querySelector(".search-progress-stage");
     const status = progress.querySelector(".search-progress-status");
@@ -140,12 +176,19 @@
       }));
       renderStatus(progress, job);
 
+      if (job.status === "cancelled") {
+        activeCancelUrl = null;
+        showCancelled(form, progress);
+        return;
+      }
       if (job.result_url) {
+        activeCancelUrl = null;
         window.location.assign(job.result_url);
         return;
       }
       window.setTimeout(() => pollJob(form, progress, statusUrl), POLL_INTERVAL_MS);
     } catch (error) {
+      cancelActiveSearch();
       showError(form, progress, error.message);
     }
   }
@@ -177,6 +220,7 @@
         credentials: "same-origin",
         headers: { Accept: "application/json" },
       }));
+      activeCancelUrl = job.cancel_url;
       pollJob(form, progress, job.status_url);
     } catch (error) {
       showError(form, progress, error.message);
@@ -193,4 +237,6 @@
       });
     });
   });
+
+  window.addEventListener("pagehide", cancelActiveSearch);
 })();

application_neo4j/translations.py

@@ -119,6 +119,7 @@ TRANSLATIONS = {
         "search.progress_stage_highlights": "Calcul des modèles importants…",
         "search.progress_stage_completed": "Recherche terminée.",
         "search.progress_stage_failed": "La recherche a échoué.",
+        "search.progress_stage_cancelled": "Recherche annulée.",
         "search.progress_elapsed": "Temps écoulé : {seconds} s",
         "search.progress_items": "Éléments traités par le serveur : {completed}/{total}",
         "search.progress_server_percent": "Progression indiquée par le serveur : {percent} %",
@@ -347,6 +348,7 @@ TRANSLATIONS = {
         "search.progress_stage_highlights": "Computing important models…",
         "search.progress_stage_completed": "Search completed.",
         "search.progress_stage_failed": "The search failed.",
+        "search.progress_stage_cancelled": "Search cancelled.",
         "search.progress_elapsed": "Elapsed time: {seconds} s",
         "search.progress_items": "Items processed by the server: {completed}/{total}",
         "search.progress_server_percent": "Progress reported by the server: {percent}%",