Synapse workers work

9 files changed, 119 insertions, 63 deletions

diff --git a/host/Rory-nginx/services/matrix/synapse/synapse-main.nix b/host/Rory-nginx/services/matrix/synapse/synapse-main.nix
index 543e782..4dc13a8 100755
--- a/host/Rory-nginx/services/matrix/synapse/synapse-main.nix
+++ b/host/Rory-nginx/services/matrix/synapse/synapse-main.nix
@@ -1,14 +1,6 @@
 { config, pkgs, lib, ... }:
 
-let
-  #federationSenders = lib.range 0 31;
-  #federationReceivers = lib.range 10000 10000;
-  #initialSyncWorkers = lib.range 10100 10100;
-  #syncWorkers = lib.range 10150 10150;
-  #streamWriters = lib.range 10200 10200;
-in
 {
-
   # Worker plumbing examples: https://github.com/element-hq/synapse/blob/master/docker/configure_workers_and_start.py
   # Documentation: https://github.com/element-hq/synapse/blob/develop/docs/workers.md
   imports = [
@@ -23,6 +15,7 @@ in
     pushers = 2;
     enableStreamWriters = false;
     enableAppserviceWorker = true;
+    enableBackgroundWorker = true;
     #eventStreamWriters = 8;
 
     # https://matrix-org.github.io/synapse/latest/usage/configuration/config_documentation.html
@@ -61,7 +54,7 @@ in
         }
         { 
           type = "http";
-          path = "/run/matrix-synapse/replication-listener.sock";
+          path = "/run/matrix-synapse/main.sock";
           resources = [ {
             names = [ "replication" ];
             compress = false;
@@ -117,7 +110,7 @@ in
 
       instance_map = {
         main = { # replication listener
-          path = "/run/matrix-synapse/replication-listener.sock";
+          path = "/run/matrix-synapse/main.sock";
         };
       };
     }
diff --git a/host/Rory-nginx/services/matrix/synapse/workers/appservice.nix b/host/Rory-nginx/services/matrix/synapse/workers/appservice.nix
index 933642b..5f1323c 100644
--- a/host/Rory-nginx/services/matrix/synapse/workers/appservice.nix
+++ b/host/Rory-nginx/services/matrix/synapse/workers/appservice.nix
@@ -18,7 +18,18 @@ in
     workers = {
       appservice = {
         worker_app = "synapse.app.generic_worker";
-        worker_listeners = [ ];
+        worker_listeners = [
+          {
+            type = "http";
+            path = "/run/matrix-synapse/appservice.sock";
+            resources = [
+              {
+                names = ["replication"];
+                compress = false;
+              }
+            ];
+          }
+        ];
       };
     };
   };
diff --git a/host/Rory-nginx/services/matrix/synapse/workers/background.nix b/host/Rory-nginx/services/matrix/synapse/workers/background.nix
new file mode 100644
index 0000000..d267124
--- /dev/null
+++ b/host/Rory-nginx/services/matrix/synapse/workers/background.nix
@@ -0,0 +1,36 @@
+{ config, pkgs, lib, ... }:
+
+let 
+  cfg = config.services.matrix-synapse;
+in
+{
+  services.matrix-synapse = lib.mkIf cfg.enableBackgroundWorker {
+    settings = {
+      instance_map = {
+        appservice = {
+          path = "/run/matrix-synapse/background.sock";
+        };
+      };
+
+      run_background_tasks_on = "background";
+    };
+
+    workers = {
+      appservice = {
+        worker_app = "synapse.app.generic_worker";
+        worker_listeners = [
+          {
+            type = "http";
+            path = "/run/matrix-synapse/background.sock";
+            resources = [
+              {
+                names = ["replication"];
+                compress = false;
+              }
+            ];
+          }
+        ];
+      };
+    };
+  };
+}
\ No newline at end of file
diff --git a/host/Rory-nginx/services/matrix/synapse/workers/federation-sender.nix b/host/Rory-nginx/services/matrix/synapse/workers/federation-sender.nix
index 9c5ed7c..c90fe3e 100644
--- a/host/Rory-nginx/services/matrix/synapse/workers/federation-sender.nix
+++ b/host/Rory-nginx/services/matrix/synapse/workers/federation-sender.nix
@@ -9,13 +9,25 @@ in
       # Instance map is not needed for federation senders
       send_federation = false;
       federation_sender_instances = lib.map (index: "federation_sender-${toString index}") federationSenders;
+      outbound_federation_restricted_to = lib.map (index: "federation_sender-${toString index}") federationSenders;
     };
 
     workers = lib.listToAttrs (lib.map (index: {
       name = "federation_sender-${toString index}";
       value = {
         worker_app = "synapse.app.generic_worker";
-        worker_listeners = [ ];
+        worker_listeners = [
+          {
+            type = "http";
+            path = "/run/matrix-synapse/pusher-${toString index}.sock";
+            resources = [
+              {
+                names = ["replication"];
+                compress = false;
+              }
+            ];
+          }
+        ];
       };
     }) federationSenders);
   };
diff --git a/host/Rory-nginx/services/matrix/synapse/workers/module.nix b/host/Rory-nginx/services/matrix/synapse/workers/module.nix
index 5193c72..9cc3d92 100644
--- a/host/Rory-nginx/services/matrix/synapse/workers/module.nix
+++ b/host/Rory-nginx/services/matrix/synapse/workers/module.nix
@@ -14,6 +14,7 @@ in
     ./federation-sender.nix
     ./appservice.nix
     ./pusher.nix
+    ./stream-writers/event-stream-writer.nix
   ];
   options.services.matrix-synapse = lib.listToAttrs (lib.map (option: {
         name = "${option}StreamWriters";
@@ -33,6 +34,7 @@ in
       ]) // {
         enableStreamWriters = lib.mkEnableOption "Enable stream writers";
         enableAppserviceWorker = lib.mkEnableOption "Enable dedicated appservice worker";
+        enableBackgroundWorker = lib.mkEnableOption "Enable dedicated background task worker";
 
         federationSenders = lib.mkOption {
           type = lib.types.int;
diff --git a/host/Rory-nginx/services/matrix/synapse/workers/pusher.nix b/host/Rory-nginx/services/matrix/synapse/workers/pusher.nix
index 037cf55..c90561e 100644
--- a/host/Rory-nginx/services/matrix/synapse/workers/pusher.nix
+++ b/host/Rory-nginx/services/matrix/synapse/workers/pusher.nix
@@ -21,7 +21,18 @@ in
         name = "pusher-${toString index}";
         value = {
           worker_app = "synapse.app.generic_worker";
-          worker_listeners = [ ];
+          worker_listeners = [
+            {
+              type = "http";
+              path = "/run/matrix-synapse/pusher-${toString index}.sock";
+              resources = [
+                {
+                  names = ["replication"];
+                  compress = false;
+                }
+              ];
+            }
+          ];
         };
       }) pushers);
     };
diff --git a/host/Rory-nginx/services/matrix/synapse/workers/stream-writers/_.nix b/host/Rory-nginx/services/matrix/synapse/workers/stream-writers/_.nix
deleted file mode 100644
index 6b0d2c2..0000000
--- a/host/Rory-nginx/services/matrix/synapse/workers/stream-writers/_.nix
+++ /dev/null
@@ -1,50 +0,0 @@
-{ 
-  count,
-  separateEventWriters ? 1,
-  separateTypingWriters ? 1, 
-  separateToDeviceWriters ? 1, 
-  separateAccountDataWriters ? 1,
-  separateReceiptsWriters ? 1,
-  separatePresenceWriters ? 1,
-  separatePusherWriters ? 1,
-}:
-{ config, pkgs, lib, ... }:
-let 
-  eventWriters = lib.range 0 (count - 1);
-  typingWriters = lib.range 0 (count - 1);
-  deviceWriters = lib.range 0 (count - 1);
-  accountDataWriters = lib.range 0 (count - 1);
-  receiptsWriters = lib.range 0 (count - 1);
-  presenceWriters = lib.range 0 (count - 1);
-  pusherWriters = lib.range 0 (count - 1);
-in
-{
-  services.matrix-synapse = {
-    settings = {
-      instance_map = lib.listToAttrs (lib.map (port: {
-        name = "stream-writer-${toString port}";
-        value = {
-          path = "/run/matrix-synapse/stream-writer-${toString port}.sock";
-        };
-      }) federationSenders);
-
-      stream_writers = {
-        events = lib.map (port: "stream-writer-events-${toString port}") federationSenders;
-        typing = lib.map (port: "stream-writer-typing-${toString port}") typingWriters;
-        to_device = lib.map (port: "stream-writer-to_device-${toString port}") deviceWriters;
-        account_data = lib.map (port: "stream-writer-account_data-${toString port}") accountDataWriters;
-        receipts = lib.map (port: "stream-writer-receipts-${toString port}") receiptsWriters;
-        presence = lib.map (port: "stream-writer-presence-${toString port}") presenceWriters;
-        push_rules = lib.map (port: "stream-writer-push_rules-${toString port}") pusherWriters;
-      };
-    };
-
-    workers = lib.listToAttrs (lib.map (port: {
-      name = "stream-writerr-${toString port}";
-      value = {
-        worker_app = "synapse.app.generic_worker";
-        worker_listeners = [ ];
-      };
-    }) federationSenders);
-  };
-}
\ No newline at end of file
diff --git a/host/Rory-nginx/services/matrix/synapse/workers/stream-writers/module.nix b/host/Rory-nginx/services/matrix/synapse/workers/stream-writers/__OLD__module.nix
index a661911..a661911 100644
--- a/host/Rory-nginx/services/matrix/synapse/workers/stream-writers/module.nix
+++ b/host/Rory-nginx/services/matrix/synapse/workers/stream-writers/__OLD__module.nix
diff --git a/host/Rory-nginx/services/matrix/synapse/workers/stream-writers/event-stream-writer.nix b/host/Rory-nginx/services/matrix/synapse/workers/stream-writers/event-stream-writer.nix
new file mode 100644
index 0000000..af7e544
--- /dev/null
+++ b/host/Rory-nginx/services/matrix/synapse/workers/stream-writers/event-stream-writer.nix
@@ -0,0 +1,41 @@
+
+{ config, pkgs, lib, ... }:
+let
+  cfg = config.services.matrix-synapse;
+  streamWriters = lib.range 0 (cfg.eventStreamWriters - 1);
+in
+{
+  services.matrix-synapse = {
+    settings = {
+      instance_map = lib.listToAttrs (lib.map (port: {
+        name = "stream-writer-${toString port}";
+        value = {
+          path = "/run/matrix-synapse/stream-writer-${toString port}.sock";
+        };
+      }) streamWriters);
+
+      stream_writers = {
+        events = lib.map (index: "stream-writer-events-${toString index}") streamWriters;
+      };
+    };
+
+    workers = lib.listToAttrs (lib.map (index: {
+      name = "stream-writer-${toString index}";
+      value = {
+        worker_app = "synapse.app.generic_worker";
+        worker_listeners = [
+          {
+            type = "http";
+            path = "/run/matrix-synapse/stream-writer-${toString index}.sock";
+            resources = [
+              {
+                names = ["replication"];
+                compress = false;
+              }
+            ];
+          }
+        ];
+      };
+    }) streamWriters);
+  };
+}
\ No newline at end of file