refactor: align Rust and Elixir with lazy script loading

Co-authored-by: manast <95200+manast@users.noreply.github.com>

Author: 2 people

elixir/lib/bullmq/redis_connection.ex

@@ -58,15 +58,10 @@ defmodule BullMQ.RedisConnection do
 
   ## Lua Script Loading
 
-  BullMQ uses Lua scripts for atomic Redis operations. All scripts are
-  automatically loaded into Redis's script cache when the connection starts.
-  This ensures the connection is fully ready for BullMQ operations (Worker,
-  Queue, QueueEvents, etc.) before it's used.
-
-  Unlike Node.js BullMQ which uses ioredis's `defineCommand` to register scripts
-  on the client, the Elixir version loads scripts via `SCRIPT LOAD` during
-  initialization and uses `EVALSHA` for execution with automatic `EVAL` fallback
-  on `NOSCRIPT` errors (in case Redis was restarted and lost its script cache).
+  BullMQ uses Lua scripts for atomic Redis operations. Scripts execute with
+  `EVALSHA` and lazily fall back to `EVAL` on `NOSCRIPT`, matching the Node.js
+  ports. For pipelined or transactional paths where that fallback is not
+  available, BullMQ loads only the specific scripts needed just before use.
 
   ## Options
 
@@ -112,10 +107,8 @@ defmodule BullMQ.RedisConnection do
 
     case Supervisor.start_link(__MODULE__, opts, name: Pool.supervisor_name(name)) do
       {:ok, pid} ->
-        # Check Redis version before loading scripts
+        # Check Redis version before the connection is used
         check_redis_version!(name)
-        # Load scripts synchronously - connection isn't ready until scripts are loaded
-        load_scripts(name)
         {:ok, pid}
 
       error ->
@@ -228,85 +221,6 @@ defmodule BullMQ.RedisConnection do
     end
   end
 
-  # Loads all BullMQ Lua scripts into Redis script cache.
-  # Called once during initialization - scripts are cached server-side in Redis,
-  # so all pool connections can use EVALSHA to execute them efficiently.
-  #
-  # To keep startup fast (a cold `SCRIPT LOAD` of many large scripts can be slow
-  # on high-latency links), we first issue a single `SCRIPT EXISTS` call with the
-  # precomputed SHA1 of every script and then only `SCRIPT LOAD` the scripts that
-  # are not already present in the server-side cache.
-  defp load_scripts(conn) do
-    scripts =
-      Scripts.list_scripts()
-      |> Enum.map(fn script_name ->
-        case Scripts.get(script_name) do
-          {content, _key_count} -> {content, Scripts.get_sha(script_name)}
-          nil -> nil
-        end
-      end)
-      |> Enum.reject(&is_nil/1)
-
-    with {:ok, missing} <- missing_scripts(conn, scripts) do
-      load_missing_scripts(conn, missing)
-    else
-      {:error, reason} ->
-        Logger.warning(
-          "BullMQ: Failed to pre-load scripts for #{inspect(conn)}: #{inspect(reason)}. " <>
-            "Scripts will be loaded on first use via EVAL fallback."
-        )
-
-        :ok
-    end
-  end
-
-  # Returns the scripts that are not yet cached server-side, using a single
-  # SCRIPT EXISTS round trip. If any SHA is missing (or the check fails), the
-  # corresponding script is treated as not loaded.
-  defp missing_scripts(conn, scripts) do
-    shas = Enum.map(scripts, fn {_content, sha} -> sha end)
-
-    case command(conn, ["SCRIPT", "EXISTS" | shas]) do
-      {:ok, existing} ->
-        missing =
-          scripts
-          |> Enum.zip(Stream.concat(existing, Stream.cycle([0])))
-          |> Enum.reject(fn {_script, loaded} -> loaded == 1 end)
-          |> Enum.map(fn {{content, _sha}, _loaded} -> content end)
-
-        {:ok, missing}
-
-      {:error, reason} ->
-        {:error, reason}
-    end
-  end
-
-  defp load_missing_scripts(conn, []) do
-    Logger.debug("BullMQ: All Lua scripts already cached in Redis for #{inspect(conn)}")
-    :ok
-  end
-
-  defp load_missing_scripts(conn, missing) do
-    commands = Enum.map(missing, fn content -> ["SCRIPT", "LOAD", content] end)
-
-    case pipeline(conn, commands) do
-      {:ok, _shas} ->
-        Logger.debug(
-          "BullMQ: Loaded #{length(commands)} Lua scripts into Redis cache for #{inspect(conn)}"
-        )
-
-        :ok
-
-      {:error, reason} ->
-        Logger.warning(
-          "BullMQ: Failed to pre-load scripts for #{inspect(conn)}: #{inspect(reason)}. " <>
-            "Scripts will be loaded on first use via EVAL fallback."
-        )
-
-        :ok
-    end
-  end
-
   @doc """
   Executes a Redis command.
 

elixir/test/bullmq/script_loading_integration_test.exs

@@ -30,7 +30,7 @@ defmodule BullMQ.ScriptLoadingIntegrationTest do
     conn_name
   end
 
-  test "loads every script into the server-side cache when it is empty" do
+  test "does not preload scripts when the connection starts" do
     flush_script_cache()
 
     conn = start_pool()
@@ -39,22 +39,24 @@ defmodule BullMQ.ScriptLoadingIntegrationTest do
     {:ok, existing} = RedisConnection.command(conn, ["SCRIPT", "EXISTS" | shas])
 
     assert length(existing) == length(shas)
-    assert Enum.all?(existing, &(&1 == 1))
+    assert Enum.all?(existing, &(&1 == 0))
   end
 
-  test "leaves all scripts cached when starting with a warm script cache" do
+  test "loads only the requested scripts for pipelined operations" do
     flush_script_cache()
 
-    # Warm the cache with a first connection.
-    _first = start_pool()
-
-    # A second connection must find the scripts already cached (only a single
-    # SCRIPT EXISTS round trip is needed) and must not remove or corrupt them.
-    second = start_pool()
-
-    shas = all_script_shas()
-    {:ok, existing} = RedisConnection.command(second, ["SCRIPT", "EXISTS" | shas])
-
-    assert Enum.all?(existing, &(&1 == 1))
+    conn = start_pool()
+    :ok = Scripts.ensure_scripts_loaded(conn, [:add_standard_job])
+
+    {:ok, [add_standard_job, add_parent_job]} =
+      RedisConnection.command(conn, [
+        "SCRIPT",
+        "EXISTS",
+        Scripts.get_sha(:add_standard_job),
+        Scripts.get_sha(:add_parent_job)
+      ])
+
+    assert add_standard_job == 1
+    assert add_parent_job == 0
   end
 end

rust/src/flow_producer.rs

@@ -301,6 +301,10 @@ impl FlowProducer {
         validate_flow_queue_names(&flow)?;
 
         let mut conn = self.conn.conn();
+        self.conn
+            .scripts()
+            .ensure_loaded(&mut conn, &Self::pipeline_script_names(&flow))
+            .await?;
 
         let mut pipe = redis::pipe();
         pipe.atomic();
@@ -354,6 +358,11 @@ impl FlowProducer {
         }
 
         let mut conn = self.conn.conn();
+        let script_names: Vec<&str> = flows.iter().flat_map(Self::pipeline_script_names).collect();
+        self.conn
+            .scripts()
+            .ensure_loaded(&mut conn, &script_names)
+            .await?;
 
         let mut pipe = redis::pipe();
         pipe.atomic();
@@ -701,6 +710,39 @@ impl FlowProducer {
         })
     }
 
+    fn pipeline_script_names(node: &FlowJob) -> Vec<&'static str> {
+        let mut names = Vec::new();
+        Self::collect_pipeline_script_names(node, &mut names);
+        names
+    }
+
+    fn collect_pipeline_script_names(node: &FlowJob, names: &mut Vec<&'static str>) {
+        if let Some(children) = &node.children {
+            if !children.is_empty() {
+                names.push("addParentJob");
+                for child in children {
+                    Self::collect_pipeline_script_names(child, names);
+                }
+                return;
+            }
+        }
+
+        let script_name = if node.opts.as_ref().and_then(|opts| opts.delay).unwrap_or(0) > 0 {
+            "addDelayedJob"
+        } else if node
+            .opts
+            .as_ref()
+            .and_then(|opts| opts.priority)
+            .unwrap_or(0)
+            > 0
+        {
+            "addPrioritizedJob"
+        } else {
+            "addStandardJob"
+        };
+        names.push(script_name);
+    }
+
     /// Add a parent job to the pipeline using the addParentJob Lua script.
     fn add_parent_job_to_pipe(
         &self,

rust/src/redis_connection.rs

@@ -40,7 +40,7 @@ fn build_client(opts: &RedisConnectionOptions, url: &str) -> Result<Client, Erro
     Ok(Client::build_with_tls(url, tls_certs)?)
 }
 
-/// A managed Redis connection that handles reconnection and script loading.
+/// A managed Redis connection that handles reconnection and script execution.
 ///
 /// This is designed to be cheaply cloneable (Arc-wrapped internals).
 /// `MultiplexedConnection` is internally multiplexed via channels,
@@ -62,8 +62,7 @@ impl RedisConnection {
         let url = opts.effective_url();
         let client = build_client(opts, &url)?;
         let scripts = ScriptRegistry::new();
-        let mut conn = client.get_multiplexed_async_connection().await?;
-        scripts.load_all(&mut conn).await?;
+        let conn = client.get_multiplexed_async_connection().await?;
 
         let inner = Arc::new(Inner {
             client,