feat(wip;sender): Centralise status registries

.forgejo/actions/setup-rust/action.yml

@@ -17,7 +17,7 @@ inputs:
     required: false
     default: ''
   rust-version:
-    description: 'Rust version to install (e.g. nightly). Defaults to 1.87.0'
+    description: 'Rust version to install (e.g. nightly). Defaults to the version specified in rust-toolchain.toml'
     required: false
     default: ''
   sccache-cache-limit:
@@ -59,9 +59,20 @@ runs:
         mkdir -p "${{ github.workspace }}/target"
         mkdir -p "${{ github.workspace }}/.rustup"
 
-    - name: Start cache restore group
+    - name: Start registry/toolchain restore group
       shell: bash
-      run: echo "::group::📦 Restoring caches (registry, toolchain, build artifacts)"
+      run: echo "::group::📦 Restoring registry and toolchain caches"
+
+    - name: Cache toolchain binaries
+      id: toolchain-cache
+      uses: actions/cache@v4
+      with:
+        path: |
+          .cargo/bin
+          .rustup/toolchains
+          .rustup/update-hashes
+        # Shared toolchain cache across all Rust versions
+        key: continuwuity-toolchain-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}
 
     - name: Cache Cargo registry and git
       id: registry-cache
@@ -77,58 +88,13 @@ runs:
         restore-keys: |
           continuwuity-cargo-registry-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-
 
-    - name: Cache toolchain binaries
-      id: toolchain-cache
-      uses: actions/cache@v4
-      with:
-        path: |
-          .cargo/bin
-          .rustup/toolchains
-          .rustup/update-hashes
-        # Shared toolchain cache across all Rust versions
-        key: continuwuity-toolchain-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}
-
-
-    - name: Setup sccache
-      uses: https://git.tomfos.tr/tom/sccache-action@v1
-
-    - name: Cache dependencies
-      id: deps-cache
-      uses: actions/cache@v4
-      with:
-        path: |
-          target/**/.fingerprint
-          target/**/deps
-          target/**/*.d
-          target/**/.cargo-lock
-          target/**/CACHEDIR.TAG
-          target/**/.rustc_info.json
-          /timelord/
-        # Dependencies cache - based on Cargo.lock, survives source code changes
-        key: >-
-          continuwuity-deps-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-${{ inputs.rust-version }}${{ inputs.cache-key-suffix && format('-{0}', inputs.cache-key-suffix) || '' }}-${{ hashFiles('rust-toolchain.toml', '**/Cargo.lock') }}
-        restore-keys: |
-          continuwuity-deps-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-${{ inputs.rust-version }}${{ inputs.cache-key-suffix && format('-{0}', inputs.cache-key-suffix) || '' }}-
-
-    - name: Cache incremental compilation
-      id: incremental-cache
-      uses: actions/cache@v4
-      with:
-        path: |
-          target/**/incremental
-        # Incremental cache - based on source code changes
-        key: >-
-          continuwuity-incremental-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-${{ inputs.rust-version }}${{ inputs.cache-key-suffix && format('-{0}', inputs.cache-key-suffix) || '' }}-${{ hashFiles('rust-toolchain.toml', '**/Cargo.lock') }}-${{ hashFiles('**/*.rs', '**/Cargo.toml') }}
-        restore-keys: |
-          continuwuity-incremental-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-${{ inputs.rust-version }}${{ inputs.cache-key-suffix && format('-{0}', inputs.cache-key-suffix) || '' }}-${{ hashFiles('rust-toolchain.toml', '**/Cargo.lock') }}-
-          continuwuity-incremental-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-${{ inputs.rust-version }}${{ inputs.cache-key-suffix && format('-{0}', inputs.cache-key-suffix) || '' }}-
-
-    - name: End cache restore group
+    - name: End registry/toolchain restore group
       shell: bash
       run: echo "::endgroup::"
 
     - name: Setup Rust toolchain
       shell: bash
+      id: rust-setup
       run: |
         # Install rustup if not already cached
         if ! command -v rustup &> /dev/null; then
@@ -156,8 +122,68 @@ runs:
           echo "::group::📦 Setting up Rust from rust-toolchain.toml"
           rustup show
         fi
+
+        RUST_VERSION=$(rustc --version | cut -d' ' -f2)
+        echo "version=$RUST_VERSION" >> $GITHUB_OUTPUT
+
         echo "::endgroup::"
 
+    - name: Install Rust components
+      if: inputs.rust-components != ''
+      shell: bash
+      run: |
+        echo "📦 Installing components: ${{ inputs.rust-components }}"
+        rustup component add ${{ inputs.rust-components }}
+
+    - name: Install Rust target
+      if: inputs.rust-target != ''
+      shell: bash
+      run: |
+        echo "📦 Installing target: ${{ inputs.rust-target }}"
+        rustup target add ${{ inputs.rust-target }}
+
+    - name: Start build cache restore group
+      shell: bash
+      run: echo "::group::📦 Restoring build cache"
+
+    - name: Setup sccache
+      uses: https://git.tomfos.tr/tom/sccache-action@v1
+
+    - name: Cache dependencies
+      id: deps-cache
+      uses: actions/cache@v4
+      with:
+        path: |
+          target/**/.fingerprint
+          target/**/deps
+          target/**/*.d
+          target/**/.cargo-lock
+          target/**/CACHEDIR.TAG
+          target/**/.rustc_info.json
+          /timelord/
+        # Dependencies cache - based on Cargo.lock, survives source code changes
+        key: >-
+          continuwuity-deps-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-${{ steps.rust-setup.outputs.version }}${{ inputs.cache-key-suffix && format('-{0}', inputs.cache-key-suffix) || '' }}-${{ hashFiles('rust-toolchain.toml', '**/Cargo.lock') }}
+        restore-keys: |
+          continuwuity-deps-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-${{ steps.rust-setup.outputs.version }}${{ inputs.cache-key-suffix && format('-{0}', inputs.cache-key-suffix) || '' }}-
+
+    - name: Cache incremental compilation
+      id: incremental-cache
+      uses: actions/cache@v4
+      with:
+        path: |
+          target/**/incremental
+        # Incremental cache - based on source code changes
+        key: >-
+          continuwuity-incremental-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-${{ steps.rust-setup.outputs.version }}${{ inputs.cache-key-suffix && format('-{0}', inputs.cache-key-suffix) || '' }}-${{ hashFiles('rust-toolchain.toml', '**/Cargo.lock') }}-${{ hashFiles('**/*.rs', '**/Cargo.toml') }}
+        restore-keys: |
+          continuwuity-incremental-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-${{ steps.rust-setup.outputs.version }}${{ inputs.cache-key-suffix && format('-{0}', inputs.cache-key-suffix) || '' }}-${{ hashFiles('rust-toolchain.toml', '**/Cargo.lock') }}-
+          continuwuity-incremental-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-${{ steps.rust-setup.outputs.version }}${{ inputs.cache-key-suffix && format('-{0}', inputs.cache-key-suffix) || '' }}-
+
+    - name: End build cache restore group
+      shell: bash
+      run: echo "::endgroup::"
+
     - name: Configure PATH and install tools
       shell: bash
       env:
@@ -211,27 +237,9 @@ runs:
           echo "CARGO_INCREMENTAL_GC_THRESHOLD=5" >> $GITHUB_ENV
         fi
 
-    - name: Install Rust components
-      if: inputs.rust-components != ''
-      shell: bash
-      run: |
-        echo "📦 Installing components: ${{ inputs.rust-components }}"
-        rustup component add ${{ inputs.rust-components }}
-
-    - name: Install Rust target
-      if: inputs.rust-target != ''
-      shell: bash
-      run: |
-        echo "📦 Installing target: ${{ inputs.rust-target }}"
-        rustup target add ${{ inputs.rust-target }}
-
     - name: Output version and summary
-      id: rust-setup
       shell: bash
       run: |
-        RUST_VERSION=$(rustc --version | cut -d' ' -f2)
-        echo "version=$RUST_VERSION" >> $GITHUB_OUTPUT
-
         echo "📋 Setup complete:"
         echo "  Rust: $(rustc --version)"
         echo "  Cargo: $(cargo --version)"

Cargo.lock

@@ -1169,6 +1169,7 @@ dependencies = [
  "conduwuit_database",
  "const-str",
  "ctor",
+ "dashmap",
  "either",
  "futures",
  "hickory-resolver",
@@ -1525,6 +1526,20 @@ version = "2.1.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "817fa642fb0ee7fe42e95783e00e0969927b96091bdd4b9b1af082acd943913b"
 
+[[package]]
+name = "dashmap"
+version = "6.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5041cc499144891f3790297212f32a74fb938e5136a14943f338ef9e0ae276cf"
+dependencies = [
+ "cfg-if",
+ "crossbeam-utils",
+ "hashbrown 0.14.5",
+ "lock_api",
+ "once_cell",
+ "parking_lot_core",
+]
+
 [[package]]
 name = "data-encoding"
 version = "2.9.0"
@@ -2112,6 +2127,12 @@ version = "0.1.2+12"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "647deb1583b14d160f85f3ff626f20b6edd366e3852c9843b06077388f794cb6"
 
+[[package]]
+name = "hashbrown"
+version = "0.14.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1"
+
 [[package]]
 name = "hashbrown"
 version = "0.15.5"

Cargo.toml

@@ -554,9 +554,6 @@ version = "0.12.0"
 default-features = false
 features = ["sync", "tls-rustls", "rustls-provider"]
 
-[workspace.dependencies.resolv-conf]
-version = "0.7.5"
-
 #
 # Patches
 #

docker/Dockerfile

@@ -48,7 +48,7 @@ EOF
 
 # Developer tool versions
 # renovate: datasource=github-releases depName=cargo-bins/cargo-binstall
-ENV BINSTALL_VERSION=1.15.11
+ENV BINSTALL_VERSION=1.16.0
 # renovate: datasource=github-releases depName=psastras/sbom-rs
 ENV CARGO_SBOM_VERSION=0.9.1
 # renovate: datasource=crate depName=lddtree

docker/musl.Dockerfile

@@ -18,7 +18,7 @@ RUN --mount=type=cache,target=/etc/apk/cache apk add \
 
 # Developer tool versions
 # renovate: datasource=github-releases depName=cargo-bins/cargo-binstall
-ENV BINSTALL_VERSION=1.15.11
+ENV BINSTALL_VERSION=1.16.0
 # renovate: datasource=github-releases depName=psastras/sbom-rs
 ENV CARGO_SBOM_VERSION=0.9.1
 # renovate: datasource=crate depName=lddtree

src/service/Cargo.toml

@@ -121,6 +121,7 @@ blurhash.workspace = true
 blurhash.optional = true
 recaptcha-verify = { version = "0.1.5", default-features = false }
 ctor.workspace = true
+dashmap = "6.1.0"
 
 [target.'cfg(all(unix, target_os = "linux"))'.dependencies]
 sd-notify.workspace = true

src/service/sending/mod.rs

@@ -4,10 +4,11 @@ mod dest;
 mod sender;
 
 use std::{
+	collections::HashMap,
 	fmt::Debug,
 	hash::{DefaultHasher, Hash, Hasher},
 	iter::once,
-	sync::Arc,
+	sync::{Arc, Mutex},
 };
 
 use async_trait::async_trait;
@@ -39,6 +40,7 @@ pub struct Service {
 	server: Arc<Server>,
 	services: Services,
 	channels: Vec<(loole::Sender<Msg>, loole::Receiver<Msg>)>,
+	statuses: Vec<sender::CurTransactionStatus>,
 }
 
 struct Services {
@@ -101,6 +103,7 @@ impl crate::Service for Service {
 				federation: args.depend::<federation::Service>("federation"),
 			},
 			channels: (0..num_senders).map(|_| loole::unbounded()).collect(),
+			statuses: vec![sender::CurTransactionStatus::new(); num_senders],
 		}))
 	}
 

src/service/sending/sender.rs

@@ -20,6 +20,7 @@ use conduwuit_core::{
 	},
 	warn,
 };
+use dashmap::DashMap;
 use futures::{
 	FutureExt, StreamExt,
 	future::{BoxFuture, OptionFuture},
@@ -55,7 +56,7 @@ use super::{
 };
 
 #[derive(Debug)]
-enum TransactionStatus {
+pub(crate) enum TransactionStatus {
 	Running,
 	Failed(u32, Instant), // number of times failed, time of last failure
 	Retrying(u32),        // number of times failed
@@ -65,7 +66,7 @@ type SendingError = (Destination, Error);
 type SendingResult = Result<Destination, SendingError>;
 type SendingFuture<'a> = BoxFuture<'a, SendingResult>;
 type SendingFutures<'a> = FuturesUnordered<SendingFuture<'a>>;
-type CurTransactionStatus = HashMap<Destination, TransactionStatus>;
+pub(crate) type CurTransactionStatus = DashMap<Destination, TransactionStatus>;
 
 const SELECT_PRESENCE_LIMIT: usize = 256;
 const SELECT_RECEIPT_LIMIT: usize = 256;
@@ -78,9 +79,13 @@ pub const EDU_LIMIT: usize = 100;
 impl Service {
 	#[tracing::instrument(skip(self), level = "debug")]
 	pub(super) async fn sender(self: Arc<Self>, id: usize) -> Result {
-		let mut statuses: CurTransactionStatus = CurTransactionStatus::new();
 		let mut futures: SendingFutures<'_> = FuturesUnordered::new();
 
+		let mut statuses = self
+			.statuses
+			.get_mut(id)
+			.expect("missing status for worker");
+
 		self.startup_netburst(id, &mut futures, &mut statuses)
 			.boxed()
 			.await;