diff --git a/Cargo.lock b/Cargo.lock
index 683b93c8a..3282d00de 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -779,9 +779,9 @@ dependencies = [
 
 [[package]]
 name = "concread"
-version = "0.4.2"
+version = "0.4.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f8c43d983bcbf6cbc1b24f9d9a6a6474d762c8744920984d2e0f4e93c2c3e9fa"
+checksum = "80f1d231b98d340d3b9a5d2ba3bd86cd25498ee1242d2e3a61916bf6f8f538a6"
 dependencies = [
  "ahash 0.7.6",
  "crossbeam-epoch",
diff --git a/Cargo.toml b/Cargo.toml
index 0b5705f35..fecbba5b1 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -109,7 +109,7 @@ clap_complete = "^4.4.3"
 # Forced by saffron/cron
 chrono = "^0.4.31"
 compact_jwt = { version = "^0.2.10", default-features = false }
-concread = "^0.4.1"
+concread = "^0.4.3"
 cron = "0.12.0"
 crossbeam = "0.8.1"
 criterion = "^0.5.1"
diff --git a/book/src/SUMMARY.md b/book/src/SUMMARY.md
index d164b5c50..f3ae56f4a 100644
--- a/book/src/SUMMARY.md
+++ b/book/src/SUMMARY.md
@@ -1,18 +1,19 @@
 # Kanidm
 
 - [Introduction to Kanidm](intro.md)
+
 - [Evaluation Quickstart](quickstart.md)
+
 - [Installing the Server](installing_the_server.md)
   - [Choosing a Domain Name](choosing_a_domain_name.md)
   - [Preparing for your Deployment](prepare_the_server.md)
   - [Server Configuration and Install](server_configuration.md)
   - [Platform Security Hardening](security_hardening.md)
   - [Server Updates](server_update.md)
+
 - [Client Tools](client_tools.md)
   - [Installing client tools](installing_client_tools.md)
 
-# Administration
-
 - [Administration](administrivia.md)
   - [Accounts and Groups](accounts_and_groups.md)
   - [Authentication and Credentials](authentication.md)
@@ -24,24 +25,25 @@
   - [Password Quality and Badlisting](password_quality.md)
   - [The Recycle Bin](recycle_bin.md)
 
-# Services
+- [Replication](repl/readme.md)
+  - [Planning](repl/planning.md)
+  - [Deployment](repl/deployment.md)
+  - [Administration](repl/administration.md)
 
-- [PAM and nsswitch](integrations/pam_and_nsswitch.md)
-- [SSH Key Distribution](ssh_key_dist.md)
-- [Oauth2](integrations/oauth2.md)
-- [LDAP](integrations/ldap.md)
-- [RADIUS](integrations/radius.md)
+- [Service Integrations](integrations/readme.md)
+  - [PAM and nsswitch](integrations/pam_and_nsswitch.md)
+  - [SSH Key Distribution](integrations/ssh_key_dist.md)
+  - [Oauth2](integrations/oauth2.md)
+  - [LDAP](integrations/ldap.md)
+  - [RADIUS](integrations/radius.md)
 
-# Synchronisation
+- [Service Integration Examples](examples/readme.md)
+  - [Kubernetes Ingress](examples/k8s_ingress_example.md)
+  - [Traefik](examples/traefik.md)
 
-- [Concepts](sync/concepts.md)
-- [FreeIPA](sync/freeipa.md)
-- [LDAP](sync/ldap.md)
-
-# Integration Examples
-
-- [Kubernetes Ingress](examples/k8s_ingress_example.md)
-- [Traefik](integrations/traefik.md)
+- [Synchronisation](sync/concepts.md)
+  - [FreeIPA](sync/freeipa.md)
+  - [LDAP](sync/ldap.md)
 
 # Support
 
@@ -64,8 +66,5 @@
   - [Replication Coordinator](developers/designs/replication_coord.md)
 - [Python Module](developers/python.md)
 - [RADIUS Integration](developers/radius.md)
-
-## Packaging
-
 - [Packaging](packaging.md)
   - [Debian/Ubuntu](packaging_debs.md)
diff --git a/book/src/authentication.md b/book/src/authentication.md
index 8bcaec802..516f6bd33 100644
--- a/book/src/authentication.md
+++ b/book/src/authentication.md
@@ -44,11 +44,13 @@ These processes are very similar. You can send a credential reset link to a user
 directly enroll their own credentials. To generate this link or qrcode:
 
 ```bash
+kanidm person credential create-reset-token <account_id> [<time to live in seconds>]
 kanidm person credential create-reset-token demo_user --name idm_admin
+kanidm person credential create-reset-token demo_user 86400 --name idm_admin
 # The person can use one of the following to allow the credential reset
-# 
+#
 # Scan this QR Code:
-# 
+#
 # █████████████████████████████████████████████
 # █████████████████████████████████████████████
 # ████ ▄▄▄▄▄ █▄██ ▀▀▀▄▀▀█ ▄▀▀▀▀▄▀▀▄█ ▄▄▄▄▄ ████
@@ -72,7 +74,7 @@ kanidm person credential create-reset-token demo_user --name idm_admin
 # ████▄▄▄▄▄▄▄█▄█▄▄▄▄▄▄█▄█▄██▄█▄▄▄█▄██▄███▄▄████
 # █████████████████████████████████████████████
 # ▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀
-# 
+#
 # This link: https://localhost:8443/ui/reset?token=8qDRG-AE1qC-zjjAT-0Fkd6
 # Or run this command: kanidm person credential use-reset-token 8qDRG-AE1qC-zjjAT-0Fkd6
 ```
@@ -80,8 +82,9 @@ kanidm person credential create-reset-token demo_user --name idm_admin
 If the user wishes you can direct them to `https://idm.mydomain.name/ui/reset` where they can
 manually enter their token value.
 
-Each token can be used only once within a 24 hour period. Once the credentials have been set the
-token is immediately invalidated.
+Once the credential has been set the token is immediately invalidated and can never be used again.
+By default the token is valid for 1 hour. You can request a longer token validity time when creating
+the token. Tokens are only allowed to be valid for a maximum of 24 hours.
 
 ### Resetting Credentials Directly
 
@@ -97,14 +100,14 @@ kanidm person credential update demo_user --name idm_admin
 # uuid: 0e19cd08-f943-489e-8ff2-69f9eacb1f31
 # generated password: set
 # Can Commit: true
-# 
+#
 # cred update (? for help) # : pass
-# New password: 
+# New password:
 # New password: [hidden]
-# Confirm password: 
+# Confirm password:
 # Confirm password: [hidden]
 # success
-# 
+#
 # cred update (? for help) # : commit
 # Do you want to commit your changes? yes
 # success
@@ -112,6 +115,9 @@ kanidm login --name demo_user
 kanidm self whoami --name demo_user
 ```
 
+{{#template templates/kani-warning.md imagepath=images title=Warning! text=Don't use the direct
+credential reset to lock or invalidate an account. You should expire the account instead. }}
+
 ## Reauthentication / Privilege Access Mode
 
 To allow for longer lived sessions in Kanidm, by default sessions are issued in a "privilege
@@ -133,5 +139,6 @@ To reauthenticate
 kanidm reauth -D william
 ```
 
-> **NOTE** During reauthentication can only use the same credential that was used to initially
-> authenticate to the session. The reauth flow will not allow any other credentials to be used!
+> **NOTE** During reauthentication an account must use the same credential that was used to
+> initially authenticate to the session. The reauth flow will not allow any other credentials to be
+> used!
diff --git a/book/src/database_maint.md b/book/src/database_maint.md
index 0597aca29..96041d047 100644
--- a/book/src/database_maint.md
+++ b/book/src/database_maint.md
@@ -6,21 +6,21 @@ In some (rare) cases you may need to reindex. Please note the server will someti
 startup as a result of the project changing its internal schema definitions. This is normal and
 expected - you may never need to start a reindex yourself as a result!
 
-You'll likely notice a need to reindex if you add indexes to schema and you see a message in your
-logs such as:
+You only need to reindex if you add custom schema elements and you see a message in your logs such
+as:
 
 ```
 Index EQUALITY name not found
 Index {type} {attribute} not found
 ```
 
-This indicates that an index of type equality has been added for name, but the indexing process has
-not been run. The server will continue to operate and the query execution code will correctly
+This indicates that an index of type equality has been added for `name`, but the indexing process
+has not been run. The server will continue to operate and the query execution code will correctly
 process the query - however it will not be the optimal method of delivering the results as we need
 to disregard this part of the query and act as though it's un-indexed.
 
 Reindexing will resolve this by forcing all indexes to be recreated based on their schema
-definitions (this works even though the schema is in the same database!)
+definitions.
 
 ```bash
 docker stop <container name>
@@ -29,8 +29,6 @@ docker run --rm -i -t -v kanidmd:/data \
 docker start <container name>
 ```
 
-Generally, reindexing is a rare action and should not normally be required.
-
 ## Vacuum
 
 Vacuuming is the process of reclaiming un-used pages from the database freelists, as well as
diff --git a/book/src/examples/readme.md b/book/src/examples/readme.md
new file mode 100644
index 000000000..4ab803cf8
--- /dev/null
+++ b/book/src/examples/readme.md
@@ -0,0 +1,6 @@
+# Service Integration Examples
+
+This chapter demonstrates examples of services and their configuration to integrate with Kanidm.
+
+If you wish to contribute more examples, please open a PR in the
+[Kanidm Project Book](https://github.com/kanidm/kanidm/tree/master/book/src/examples).
diff --git a/book/src/integrations/traefik.md b/book/src/examples/traefik.md
similarity index 100%
rename from book/src/integrations/traefik.md
rename to book/src/examples/traefik.md
diff --git a/book/src/installing_client_tools.md b/book/src/installing_client_tools.md
index a465ef39e..f74fbba82 100644
--- a/book/src/installing_client_tools.md
+++ b/book/src/installing_client_tools.md
@@ -1,15 +1,15 @@
 # Installing Client Tools
 
-> **NOTE** As this project is in a rapid development phase, running different release versions will
-> likely present incompatibilities. Ensure you're running matching release versions of client and
-> server binaries. If you have any issues, check that you are running the latest software.
+> **NOTE** Running different release versions will likely present incompatibilities. Ensure you're
+> running matching release versions of client and server binaries. If you have any issues, check
+> that you are running the latest version of Kanidm.
 
 ## From packages
 
 Kanidm currently is packaged for the following systems:
 
 - OpenSUSE Tumbleweed
-- OpenSUSE Leap 15.4
+- OpenSUSE Leap 15.4/15.5/15.6
 - MacOS
 - Arch Linux
 - NixOS
@@ -18,16 +18,16 @@ Kanidm currently is packaged for the following systems:
 
 The `kanidm` client has been built and tested from Windows, but is not (yet) packaged routinely.
 
-### OpenSUSE Tumbleweed
+### OpenSUSE Tumbleweed / Leap 15.6
 
-Kanidm has been part of OpenSUSE Tumbleweed since October 2020. You can install the clients with:
+Kanidm is available in Tumbleweed and Leap 15.6. You can install the clients with:
 
 ```bash
 zypper ref
 zypper in kanidm-clients
 ```
 
-### OpenSUSE Leap 15.3/15.4
+### OpenSUSE Leap 15.4/15.5
 
 Using zypper you can add the Kanidm leap repository with:
 
@@ -88,22 +88,12 @@ You can then install with:
 dnf install kanidm-clients
 ```
 
-## Cargo
-
-The tools are available as a cargo download if you have a rust tool chain available. To install rust
-you should follow the documentation for [rustup](https://rustup.rs/). These will be installed into
-your home directory. To update these, re-run the install command.
-
-```bash
-cargo install kanidm_tools
-```
-
 ## Tools Container
 
 In some cases if your distribution does not have native kanidm-client support, and you can't access
 cargo for the install for some reason, you can use the cli tools from a docker container instead.
 
-This really is a "last resort" and we don't really recommend this for day to day usage.
+This is a "last resort" and we don't really recommend this for day to day usage.
 
 ```bash
 echo '{}' > ~/.cache/kanidm_tokens
@@ -126,6 +116,16 @@ If you have a ca.pem you may need to bind mount this in as required as well.
 alias kanidm="docker run ..."
 ```
 
+## Cargo
+
+The tools are available as a cargo download if you have a rust tool chain available. To install rust
+you should follow the documentation for [rustup](https://rustup.rs/). These will be installed into
+your home directory. To update these, re-run the install command.
+
+```bash
+cargo install kanidm_tools
+```
+
 ## Initializing the configuration
 
 The client requires a configuration file to connect to the server. This should be at
diff --git a/book/src/integrations/ldap.md b/book/src/integrations/ldap.md
index c1317fce5..c6b3f68d6 100644
--- a/book/src/integrations/ldap.md
+++ b/book/src/integrations/ldap.md
@@ -11,7 +11,7 @@ interface for these legacy applications and services.
 {{#template ../templates/kani-warning.md
 imagepath=../images
 title=Warning!
-text=The LDAP server in Kanidm is not a fully RFC-compliant LDAP server. This is intentional, as Kanidm wants to cover the common use cases - simple bind and search.
+text=The LDAP server in Kanidm is not a full LDAP server. This is intentional, as Kanidm wants to cover the common use cases - simple bind and search. The parts we do support are RFC compliant however.
 }}
 
 <!-- deno-fmt-ignore-end -->
@@ -26,10 +26,10 @@ IDM just like Kanidm!
 
 ## Data Mapping
 
-Kanidm cannot be mapped 100% to LDAP's objects. This is because LDAP types are simple key-values on
-objects which are all UTF8 strings (or subsets thereof) based on validation (matching) rules. Kanidm
-internally implements complex data types such as tagging on SSH keys, or multi-value credentials.
-These can not be represented in LDAP.
+Kanidm entries cannot be mapped 100% to LDAP's objects. This is because LDAP types are simple
+key-values on objects which are all UTF8 strings (or subsets thereof) based on validation (matching)
+rules. Kanidm internally implements complex structured data types such as tagging on SSH keys, or
+multi-value credentials. These can not be represented in LDAP.
 
 Many of the structures in Kanidm do not correlate closely to LDAP. For example Kanidm only has a GID
 number, where LDAP's schemas define both a UID number and a GID number.
@@ -52,11 +52,12 @@ limitation for the consuming applications.
 
 ## Security
 
-### TLS
+### LDAPS vs StartTLS
 
-StartTLS is not supported due to security risks such as credential leakage and MITM attacks that are
-fundamental in how StartTLS works and can not be repaired. LDAPS is the only secure method of
-communicating to any LDAP server. Kanidm will use it's certificates for both HTTPS and LDAPS.
+StartTLS is _not_ supported due to security risks such as credential leakage and MITM attacks that
+are fundamental in how StartTLS works. StartTLS can not be repaired to prevent this. LDAPS is the
+only secure method of communicating to any LDAP server. Kanidm will use its certificates for both
+HTTPS and LDAPS.
 
 ### Writes
 
diff --git a/book/src/integrations/oauth2.md b/book/src/integrations/oauth2.md
index 046aa9351..2be321935 100644
--- a/book/src/integrations/oauth2.md
+++ b/book/src/integrations/oauth2.md
@@ -45,12 +45,11 @@ decisions to Kanidm.
 
 It's important for you to know _how_ your resource server supports OAuth2. For example, does it
 support RFC 7662 token introspection or does it rely on OpenID connect for identity information?
-Does the resource server support PKCE S256?
 
 In general Kanidm requires that your resource server supports:
 
 - HTTP basic authentication to the authorisation server
-- PKCE S256 code verification to prevent certain token attack classes
+- PKCE S256 code verification
 - OIDC only - JWT ES256 for token signatures
 
 Kanidm will expose its OAuth2 APIs at the following URLs:
@@ -61,16 +60,16 @@ Kanidm will expose its OAuth2 APIs at the following URLs:
 - rfc7662 token introspection url: `https://idm.example.com/oauth2/token/introspect`
 - rfc7009 token revoke url: `https://idm.example.com/oauth2/token/revoke`
 
-OpenID Connect discovery - you need to substitute your OAuth2 client id in the following urls:
+OpenID Connect discovery - you need to substitute your OAuth2 `:client_id:` in the following urls:
 
-- OpenID connect issuer uri: `https://idm.example.com/oauth2/openid/:client\_id:/`
+- OpenID connect issuer uri: `https://idm.example.com/oauth2/openid/:client_id:/`
 - OpenID connect discovery:
-  `https://idm.example.com/oauth2/openid/:client\_id:/.well-known/openid-configuration`
+  `https://idm.example.com/oauth2/openid/:client_id:/.well-known/openid-configuration`
 
 For manual OpenID configuration:
 
-- OpenID connect userinfo: `https://idm.example.com/oauth2/openid/:client\_id:/userinfo`
-- token signing public key: `https://idm.example.com/oauth2/openid/:client\_id:/public\_key.jwk`
+- OpenID connect userinfo: `https://idm.example.com/oauth2/openid/:client_id:/userinfo`
+- token signing public key: `https://idm.example.com/oauth2/openid/:client_id:/public_key.jwk`
 
 ### Scope Relationships
 
@@ -112,7 +111,7 @@ the resource server.
 ### Create the Kanidm Configuration
 
 After you have understood your resource server requirements you first need to configure Kanidm. By
-default members of "system\_admins" or "idm\_hp\_oauth2\_manage\_priv" are able to create or manage
+default members of `system_admins` or `idm_hp_oauth2_manage_priv` are able to create or manage
 OAuth2 resource server integrations.
 
 You can create a new resource server with:
@@ -134,7 +133,7 @@ kanidm system oauth2 update-scope-map nextcloud nextcloud_admins admin
 {{#template ../templates/kani-warning.md
 imagepath=../images
 title=WARNING
-text=If you are creating an OpenID Connect (OIDC) resource server you <b>MUST</b> provide a scope map named <code>openid</code>. Without this, OpenID clients <b>WILL NOT WORK</b>!
+text=If you are creating an OpenID Connect (OIDC) resource server you <b>MUST</b> provide a scope map named <code>openid</code>. Without this, OpenID Connect clients <b>WILL NOT WORK</b>!
 }}
 
 <!-- deno-fmt-ignore-end -->
@@ -182,8 +181,8 @@ kanidm system oauth2 show-basic-secret nextcloud
 
 ### Configure the Resource Server
 
-On your resource server, you should configure the client ID as the "oauth2\_rs\_name" from Kanidm,
-and the password to be the value shown in "oauth2\_rs\_basic\_secret". Ensure that the code
+On your resource server, you should configure the client ID as the `oauth2_rs_name` from Kanidm, and
+the password to be the value shown in `oauth2_rs_basic_secret`. Ensure that the code
 challenge/verification method is set to S256.
 
 You should now be able to test authorisation.
@@ -211,7 +210,8 @@ will not affect others.
 
 {{#template ../templates/kani-warning.md
 imagepath=../images
-title=WARNING text=Changing these settings MAY have serious consequences on the security of your resource server. You should avoid changing these if at all possible!
+title=WARNING
+text=Changing these settings MAY have serious consequences on the security of your resource server. You should avoid changing these if at all possible!
 }}
 
 <!-- deno-fmt-ignore-end -->
@@ -242,7 +242,8 @@ exchange. PKCE can not be disabled for public clients for this reason.
 
 {{#template ../templates/kani-warning.md
 imagepath=../images
-title=WARNING text=Public clients have many limitations compared to confidential clients. You should avoid them if possible.
+title=WARNING
+text=Public clients have many limitations compared to confidential clients. You should avoid them if possible.
 }}
 
 <!-- deno-fmt-ignore-end -->
diff --git a/book/src/integrations/pam_and_nsswitch.md b/book/src/integrations/pam_and_nsswitch.md
index e1fb60ee8..254f57ec0 100644
--- a/book/src/integrations/pam_and_nsswitch.md
+++ b/book/src/integrations/pam_and_nsswitch.md
@@ -9,7 +9,7 @@ Kanidm into accounts that can be used on the machine for various interactive tas
 Kanidm provides a UNIX daemon that runs on any client that wants to use PAM and nsswitch
 integration. The daemon can cache the accounts for users who have unreliable networks, or who leave
 the site where Kanidm is hosted. The daemon is also able to cache missing-entry responses to reduce
-network traffic and main server load.
+network traffic and Kanidm server load.
 
 Additionally, running the daemon means that the PAM and nsswitch integration libraries can be small,
 helping to reduce the attack surface of the machine. Similarly, a tasks daemon is available that can
@@ -154,7 +154,7 @@ testgroup:x:2439676479:testunix
 
 > **HINT** Remember to also create a UNIX password with something like
 > `kanidm account posix set_password --name idm_admin demo_user`. Otherwise there will be no
-> credential for the account to authenticate.
+> credential for the account to authenticate with.
 
 ## PAM
 
@@ -220,14 +220,6 @@ account    [default=1 ignore=ignore success=ok]  pam_succeed_if.so uid >= 1000 q
 account    sufficient    pam_kanidm.so ignore_unknown_user
 account    required      pam_deny.so
 
-# /etc/pam.d/common-password-pc
-# Controls flow of what happens when a user invokes the passwd command. Currently does NOT
-# interact with kanidm.
-password    [default=1 ignore=ignore success=ok] pam_localuser.so
-password    required    pam_unix.so use_authtok nullok shadow try_first_pass
-password    [default=1 ignore=ignore success=ok]  pam_succeed_if.so uid >= 1000 quiet_success quiet_fail
-password    required    pam_kanidm.so
-
 # /etc/pam.d/common-session-pc
 # Controls setup of the user session once a successful authentication and authorisation has
 # occurred.
@@ -237,7 +229,15 @@ session optional    pam_unix.so try_first_pass
 session optional    pam_umask.so
 session [default=1 ignore=ignore success=ok] pam_succeed_if.so uid >= 1000 quiet_success quiet_fail
 session optional    pam_kanidm.so
-    session optional    pam_env.so
+session optional    pam_env.so
+
+# /etc/pam.d/common-password-pc
+# Controls flow of what happens when a user invokes the passwd command. Currently does NOT
+# interact with kanidm.
+password    [default=1 ignore=ignore success=ok] pam_localuser.so
+password    required    pam_unix.so use_authtok nullok shadow try_first_pass
+password    [default=1 ignore=ignore success=ok]  pam_succeed_if.so uid >= 1000 quiet_success quiet_fail
+password    required    pam_kanidm.so
 ```
 
 > **WARNING:** Ensure that `pam_mkhomedir` or `pam_oddjobd` are _not_ present in any stage of your
@@ -246,7 +246,7 @@ session optional    pam_kanidm.so
 ### Fedora / CentOS
 
 > **WARNING:** Kanidm currently has no support for SELinux policy - this may mean you need to run
-> the daemon with permissive mode for the unconfined_service_t daemon type. To do this run:
+> the daemon with permissive mode for the `unconfined_service_t` daemon type. To do this run:
 > `semanage permissive -a unconfined_service_t`. To undo this run
 > `semanage permissive -d unconfined_service_t`.
 >
diff --git a/book/src/integrations/readme.md b/book/src/integrations/readme.md
new file mode 100644
index 000000000..4fea991b0
--- /dev/null
+++ b/book/src/integrations/readme.md
@@ -0,0 +1,4 @@
+# Service Integrations
+
+This chapter describes interfaces that Kanidm provides that allows external services and
+applications to integrate with and trust Kanidm as an authentication and identity provider.
diff --git a/book/src/ssh_key_dist.md b/book/src/integrations/ssh_key_dist.md
similarity index 99%
rename from book/src/ssh_key_dist.md
rename to book/src/integrations/ssh_key_dist.md
index 58caa0c74..db4a61477 100644
--- a/book/src/ssh_key_dist.md
+++ b/book/src/integrations/ssh_key_dist.md
@@ -98,7 +98,7 @@ The kanidm\_ssh\_authorizedkeys\_direct command is part of the kanidm-clients pa
 installed on the servers.
 
 To configure the tool, you should edit /etc/kanidm/config, as documented in
-[clients](./client_tools.md)
+[clients](../client_tools.md)
 
 You can test this is configured correctly by running:
 
diff --git a/book/src/intro.md b/book/src/intro.md
index a0d982c5e..2a1e70f15 100644
--- a/book/src/intro.md
+++ b/book/src/intro.md
@@ -10,16 +10,6 @@ The intent of the Kanidm project is to:
 - Make system, network, application and web authentication easy and accessible.
 - Secure and reliable by default, aiming for the highest levels of quality.
 
-<!-- deno-fmt-ignore-start -->
-
-{{#template templates/kani-warning.md
-imagepath=images
-title=NOTICE
-text=Kanidm is still a work in progress. Many features will evolve and change over time which may not be suitable for all users.
-}}
-
-<!-- deno-fmt-ignore-end -->
-
 ## Why do I want Kanidm?
 
 Whether you work in a business, a volunteer organisation, or are an enthusiast who manages their
diff --git a/book/src/packaging_debs.md b/book/src/packaging_debs.md
index d02336496..5ba80e472 100644
--- a/book/src/packaging_debs.md
+++ b/book/src/packaging_debs.md
@@ -9,17 +9,6 @@ This happens in Docker currently, and here's some instructions for doing it for
    repository in `~/kanidm/`.
 3. Install the required dependencies by running `./scripts/install_ubuntu_dependencies.sh`.
 4. Building packages uses make, get a list by running `make -f ./platform/debian/Makefile help`
-
-   ➜ make -f platform/debian/Makefile help
-   debs/kanidm:
-      build a .deb for the Kanidm CLI
-   debs/kanidmd:
-      build a .deb for the Kanidm daemon
-   debs/kanidm-unixd:
-      build a .deb for the Kanidm UNIX tools (PAM/NSS, unixd and related tools) and SSH tools
-   debs/all:
-      build all the debs
-
 5. So if you wanted to build the package for the Kanidm CLI, run
    `make -f ./platform/debian/Makefile debs/kanidm`.
 6. The package will be copied into the `target` directory of the repository on the docker host - not
diff --git a/book/src/posix_accounts.md b/book/src/posix_accounts.md
index 11682ab0c..ac5f722bf 100644
--- a/book/src/posix_accounts.md
+++ b/book/src/posix_accounts.md
@@ -40,11 +40,13 @@ william:x:654401105:
 
 Other systems like FreeIPA use a plugin that generates a UPG as a separate group entry on creation
 of the account. This means there are two entries for an account, and they must be kept in lock-step.
+This poses a risk of desynchronisation that can and will happen on these systems leading to possible
+issues.
 
 Kanidm does neither of these. As the GID number of the user must be unique, and a user implies the
-UPG must exist, we can generate UPG's on-demand from the account. This has a single side effect -
-that you are unable to add any members to a UPG - given the nature of a user private group, this is
-the point.
+UPG must exist, we can generate UPG's on-demand from the account. This has an important side
+effect - that you are unable to add any members to a UPG - given the nature of a user private group,
+this is the point.
 
 ### GID Number Generation
 
diff --git a/book/src/prepare_the_server.md b/book/src/prepare_the_server.md
index 1f7464244..08469d677 100644
--- a/book/src/prepare_the_server.md
+++ b/book/src/prepare_the_server.md
@@ -2,10 +2,6 @@
 
 ## Software Installation Method
 
-> **NOTE** Our preferred deployment method is in containers, and this documentation assumes you're
-> running in docker. Kanidm will alternately run as a daemon/service, and server builds are
-> available for multiple platforms if you prefer this option. You will
-
 We provide docker images for the server components. They can be found at:
 
 - <https://hub.docker.com/r/kanidm/server>
@@ -20,15 +16,18 @@ docker pull kanidm/radius:latest
 docker pull kanidm/tools:latest
 ```
 
-You may need to adjust your example commands throughout this document to suit your desired server
-type if you choose not to use docker.
+> **NOTE** Our preferred deployment method is in containers, and this documentation assumes you're
+> running in docker. Kanidm will alternately run as a daemon/service, and server builds are
+> available for multiple platforms if you prefer this option. You may need to adjust the example
+> commands throughout this document to suit your desired server type if you choose not to use
+> containers.
 
 ## Development Version
 
 If you are interested in running the latest code from development, you can do this by changing the
-docker tag to `kanidm/server:devel` or `kanidm/server:x86_64_v3_devel` instead. Many people run the
-development version, and it is extremely reliable, but occasional rough patches may occur. If you
-report issues, we will make every effort to help resolve them.
+docker tag to `kanidm/server:devel` instead. Many people run the development version, and it is
+extremely reliable, but occasional rough patches may occur. If you report issues, we will make every
+effort to help resolve them.
 
 ## System Requirements
 
@@ -39,8 +38,8 @@ Kanidm relies on modern CPU optimisations for many operations. As a result your
 - `x86_64` supporting `x86_64_v2` operations.
 - `aarch64` supporting `neon_v8` operations.
 
-Older or unsupported CPUs may raise a SIGIL (Illegal Instruction) on hardware that is not supported
-by the project.
+Older or unsupported CPUs may raise a `SIGILL` (Illegal Instruction) on hardware that is not
+supported by the project.
 
 <!-- deno-fmt-ignore-start -->
 
@@ -83,9 +82,9 @@ The key.pem should be a single PEM private key, with no encryption. The file con
 similar to:
 
 ```bash
------BEGIN RSA PRIVATE KEY-----
+-----BEGIN PRIVATE KEY-----
 MII...<base64>
------END RSA PRIVATE KEY-----
+-----END PRIVATE KEY-----
 ```
 
 The chain.pem is a series of PEM formatted certificates. The leaf certificate, or the certificate
diff --git a/book/src/quickstart.md b/book/src/quickstart.md
index 8a6b93cc8..29ab43560 100644
--- a/book/src/quickstart.md
+++ b/book/src/quickstart.md
@@ -3,7 +3,7 @@
 This section will guide you through a quick setup of Kanidm for evaluation. It's recommended that
 for a production deployment you follow the steps in the
 [installation chapter](installing_the_server.html) instead as there are a number of security
-considerations you should understand.
+considerations you should be aware of for production deployments.
 
 ### Requirements
 
diff --git a/book/src/repl/administration.md b/book/src/repl/administration.md
new file mode 100644
index 000000000..e6711b764
--- /dev/null
+++ b/book/src/repl/administration.md
@@ -0,0 +1,30 @@
+# Administration
+
+## Renew Replication Identity Certificate
+
+The replication identity certificate defaults to an expiry of 180 days.
+
+To renew this run the command:
+
+```bash
+docker exec -i -t <container name> \
+  kanidmd renew-replication-certificate
+# certificate: "MII....."
+```
+
+You must then copy the new certificate to other nodes in the topology.
+
+> **NOTE** In the future we will develop a replication coordinator so that you don't have to
+> manually renew this. But for now, if you want replication, you have to do it the hard way.
+
+## Refresh a Lagging Consumer
+
+If a consumer has been offline for more than 7 days, its error log will display that it requires a
+refresh.
+
+You can manually perform this on the affected node.
+
+```bash
+docker exec -i -t <container name> \
+  kanidmd refresh-replication-consumer
+```
diff --git a/book/src/repl/deployment.md b/book/src/repl/deployment.md
new file mode 100644
index 000000000..3565d4889
--- /dev/null
+++ b/book/src/repl/deployment.md
@@ -0,0 +1,151 @@
+# Deployment
+
+<!-- deno-fmt-ignore-start -->
+
+{{#template ../templates/kani-warning.md
+imagepath=../images
+title=WARNING
+text=Replication is a newely developed feature. This means it requires manual configuration and careful monitoring. You should take regular backups if you choose to proceed.
+}}
+
+<!-- deno-fmt-ignore-end -->
+
+## Node Setup
+
+On the servers that you wish to participate in the replication topology, you must enable replication
+in their server.toml to allow identity certificates to be generated.
+
+```toml
+# server.toml
+
+# To proceed with replication, replace the line "ACK_HERE" with
+# "i acknowledge that replication is in development = true" where the spaces
+# are replaced with '_'
+
+ACK_HERE
+
+[replication]
+# The hostname and port of the server that other nodes will connect to.
+origin = "repl://localhost:8444"
+# The bind address of the replication port.
+bindaddress = "127.0.0.1:8444"
+```
+
+Once configured, deploy this config to your servers and restart the nodes.
+
+## Manual Node Configurations
+
+> **NOTE** In the future we will develop a replication coordinator so that you don't have to
+> manually configure this. But for now, if you want replication, you have to do it the hard way.
+
+Each node has an identify certificate that is internally generated and used to communicate with
+other nodes in the topology. This certificate is also used by other nodes to validate this node.
+
+Let's assume we have two servers - A and B. We want B to consume (pull) data from A initially as A
+is our "first server".
+
+First display the identity certificate of A.
+
+```bash
+# Server A
+docker exec -i -t <container name> \
+  kanidmd show-replication-certificate
+# certificate: "MII....."
+```
+
+Now on node B, configure the replication node config.
+
+```toml
+[replication]
+# ...
+
+[replication."repl://origin_of_A:port"]
+type = "mutual-pull"
+partner_cert = "MII... <as output from A show-replication-cert>"
+```
+
+Now we must configure A to pull from B.
+
+```bash
+# Server B
+docker exec -i -t <container name> \
+  kanidmd show-replication-certificate
+# certificate: "MII....."
+```
+
+Now on node A, configure the replication node config.
+
+```toml
+[replication]
+# ...
+
+[replication."repl://origin_of_B:port"]
+type = "mutual-pull"
+partner_cert = "MII... <as output from B show-replication-cert>"
+```
+
+Then restart both servers. Initially the servers will refuse to synchronise as their databases do
+not have matching `domain_uuids`. To resolve this you can instruct B to manually refresh from A
+with:
+
+```bash
+# Server B
+docker exec -i -t <container name> \
+  kanidmd refresh-replication-consumer
+```
+
+## Partially Automated Node Configurations
+
+> **NOTE** In the future we will develop a replication coordinator so that you don't have to
+> manually configure this. But for now, if you want replication, you have to do it the hard way.
+
+This is the same as the manual process, but a single server is defined as the "primary" and the
+partner server is the "secondary". This means that if database issues occur the content of the
+primary will take precedence over the secondary. For our example we will define A as the primary and
+B as the secondary.
+
+First display the identity certificate
+
+```bash
+# Server A
+docker exec -i -t <container name> \
+  kanidmd show-replication-certificate
+# certificate: "MII....."
+```
+
+Now a secondary, configure the replication node config.
+
+```toml
+[replication]
+# ...
+
+[replication."repl://origin_of_A:port"]
+type = "mutual-pull"
+partner_cert = "MII... <as output from A show-replication-cert>"
+automatic_refresh = true
+```
+
+Now we must configure A to pull from B.
+
+```bash
+# Server B
+docker exec -i -t <container name> \
+  kanidmd show-replication-certificate
+# certificate: "MII....."
+```
+
+Now on node A, configure the replication node config. It is critical here that you do NOT set
+`automatic_refresh`.
+
+```toml
+[replication]
+# ...
+
+[replication."repl://origin_of_B:port"]
+type = "mutual-pull"
+partner_cert = "MII... <as output from B show-replication-cert>"
+# automatic_refresh = false
+```
+
+Then restart both servers. B (secondary) will automatically refresh from A (primary) and then
+replication will continue bi-directionally from that point.
diff --git a/book/src/repl/planning.md b/book/src/repl/planning.md
new file mode 100644
index 000000000..d0d8b3208
--- /dev/null
+++ b/book/src/repl/planning.md
@@ -0,0 +1,64 @@
+# Planning
+
+<!-- deno-fmt-ignore-start -->
+
+{{#template ../templates/kani-warning.md
+imagepath=../images
+title=WARNING
+text=Replication is a newely developed feature. This means it requires manual configuration and careful monitoring. You should keep backups if you choose to proceed.
+}}
+
+<!-- deno-fmt-ignore-end -->
+
+It is important that you plan your replication deployment before you proceed. You may have a need
+for high availability within a datacentre, geographic redundancy, or improvement of read scaling.
+
+## Improvement of Read Throughput
+
+Addition of replicas can improve the amount of read and authentication operations performed over the
+topology as a whole. This is because read operations throughput is additive between nodes.
+
+For example, if you had two servers that can process 1000 authentications per second each, then when
+in replication the topology can process 2000 authentications per second.
+
+However, while you may gain in read throughput, you must account for downtime - you should not
+always rely on every server to be online.
+
+The optimal loading of any server is approximately 50%. This allows overhead to absorb load if
+nearby nodes experience outages. It also allows for absorption of load spikes or other unexpected
+events.
+
+It is important to note however that as you add replicas the _write_ throughput does not increase in
+the same way as read throughput. This is because for each write that occurs on a node, it must be
+replicated and written to every other node. Therefore your write throughput is always bounded by the
+_slowest_ server in your topology. In reality there is a "slight" improvement in writes due to
+coalescing that occurs as part of replication, but you should assume that writes are not improved
+through the addition of more nodes.
+
+## Directing Clients to Live Servers
+
+Operating replicas of Kanidm allows you to minimise outages if a single or multiple servers
+experience downtime. This can assist you with patching and other administrative tasks that you must
+perform.
+
+However, there are some key limitations to this fault tolerance.
+
+You require a method to fail over between servers. This generally involves a load balancer, which
+itself must be fault tolerant. Load balancers can be made fault tolerant through the use of
+protocols like `CARP` or `VRRP`, or by configuration of routers with anycast.
+
+If you elect to use `CARP` or `VRRP` directly on your Kanidm servers, then be aware that you will be
+configuring your systems as active-passive, rather than active-active, so you will not benefit from
+improved read throughput. Contrast, anycast will always route to the closest Kanidm server and will
+failover to nearby servers so this may be an attractive choice.
+
+You should _NOT_ use DNS based failover mechanisms as clients can cache DNS records and remain
+"stuck" to a node in a failed state.
+
+## Maximum Downtime of a Server
+
+Kanidm's replication protocol enforces limits on how long a server can be offline. This is due
+to how tombstones are handled. By default the maximum is 7 days. If a server is offline for more
+than 7 days a refresh will be required for that server to continue participation in the topology.
+
+It is important you avoid extended downtime of servers to avoid this condition.
diff --git a/book/src/repl/readme.md b/book/src/repl/readme.md
new file mode 100644
index 000000000..ffd1a8c39
--- /dev/null
+++ b/book/src/repl/readme.md
@@ -0,0 +1,90 @@
+# Replication
+
+## Introduction
+
+Replication allows two or more Kanidm servers to exchange their databases and keep their content
+synchronised. This is critical to allow multiple servers to act in failover groups for highly
+available infrastructure.
+
+Kanidm replication is eventually consistent. This means that there are no elections or quorums
+required between nodes - all nodes can accept writes and distribute them to all other nodes. This is
+important for security and performance.
+
+Because replication is eventually consistent, this means that there can be small delays between
+different servers receiving a change. This may result in some users noticing discrepancies that are
+quickly resolved.
+
+To minimise this, it's recommended that when you operate replication in a highly available
+deployment that you have a load balancer that uses sticky sessions so that users are redirected to
+the same server unless a failover event occurs. This will help to minimise discrepancies.
+Alternately you can treat replication and "active-passive" and have your load balancer failover
+between the two nodes. Since replication is eventually consistent, there is no need for a 
+failover or failback procedure.
+
+In this chapter we will cover the details of planning, deploying and maintaining replication between
+Kanidm servers.
+
+## Vocabulary
+
+Replication requires us to use introduce specific words so that we can describe the replication
+environment.
+
+### Change
+
+An update made in the database.
+
+### Node
+
+A server that is participating in replication.
+
+### Pull
+
+The act of requesting data from a remote server.
+
+### Push
+
+The act of supplying data to a remote server.
+
+### Node Configuration
+
+A descriptor that allows a node to pull from another node.
+
+### Converge
+
+To approach the same database state.
+
+### Topology
+
+The collection of servers that are joined in replication and converge on the same database content.
+The topology is defined by the set of node configurations.
+
+### Replication
+
+The act of exchanging data from one node to another.
+
+### Supplier
+
+The node that is supplying data to another node.
+
+### Consumer
+
+The node that is replicating content from a supplier.
+
+### Refresh
+
+Deleting all of a consumer's database content, and replacing it with the content of a supplier.
+
+### Incremental Replication
+
+When a supplier provides a "differential" between the state of the consumer and the supplier for the
+consumer to apply.
+
+### Conflict
+
+If a consumer can not validate a change that a supplier provided, then the entry may move to a
+conflict state. All nodes will converge to the same conflict state over time.
+
+### Tombstone
+
+A marker entry that displays an entry has been deleted. This allow all servers to converge and
+delete the data.
diff --git a/book/src/security_hardening.md b/book/src/security_hardening.md
index 21c064df0..272a46746 100644
--- a/book/src/security_hardening.md
+++ b/book/src/security_hardening.md
@@ -1,8 +1,8 @@
 # Security Hardening
 
 Kanidm ships with a secure-by-default configuration, however that is only as strong as the
-environment that Kanidm operates in. This could be your container environment or your Unix-like
-system.
+environment that Kanidm operates in. This means the security of your container environment and
+server is extremely important when running Kanidm.
 
 This chapter will detail a number of warnings and security practices you should follow to ensure
 that Kanidm operates in a secure environment.
@@ -14,8 +14,8 @@ full-network take over, also known as "game over".
 The unixd resolver is also a high value target as it can be accessed to allow unauthorised access to
 a server, to intercept communications to the server, or more. This also must be protected carefully.
 
-For this reason, Kanidm's components must be protected carefully. Kanidm avoids many classic attacks
-by being developed in a memory safe language, but risks still exist.
+For this reason, Kanidm's components must be secured and audited. Kanidm avoids many classic attacks
+by being developed in a memory safe language, but risks still exist in the operating environment.
 
 ## Startup Warnings
 
@@ -36,7 +36,7 @@ WARNING: DB folder /tmp has 'everyone' permission bits in the mode. This could b
 
 Each warning highlights an issue that may exist in your environment. It is not possible for us to
 prescribe an exact configuration that may secure your system. This is why we only present possible
-risks.
+risks and you must make informed decisions on how to resolve them.
 
 ### Should be Read-only to Running UID
 
@@ -118,13 +118,6 @@ and owned by root:
 
 This file should be "everyone"-readable, which is why the bits are defined as such.
 
-> NOTE: Why do you use 440 or 444 modes?
->
-> A bug exists in the implementation of readonly() in rust that checks this as "does a write bit
-> exist for any user" vs "can the current UID write the file?". This distinction is subtle but it
-> affects the check. We don't believe this is a significant issue though, because setting these to
-> 440 and 444 helps to prevent accidental changes by an administrator anyway
-
 ## Running as Non-root in docker
 
 The commands provided in this book will run kanidmd as "root" in the container to make the
@@ -161,4 +154,4 @@ docker run --rm -i -t -u 1000:1000 -v kanidmd:/data kanidm/server:latest /sbin/k
 
 ## Minimum TLS key lengths
 
-We enforce a minimum RSA key length of 2048 bits, and EC keys need 224 bits.
+We enforce a minimum RSA key length of 2048 bits, and ECDSA keys need 224 bits.
diff --git a/book/src/server_configuration.md b/book/src/server_configuration.md
index 186a61d57..388df0165 100644
--- a/book/src/server_configuration.md
+++ b/book/src/server_configuration.md
@@ -3,7 +3,7 @@
 ## Configuring server.toml
 
 You need a configuration file in the volume named `server.toml`. (Within the container it should be
-`/data/server.toml`) Its contents should be as follows:
+`/data/server.toml`) The following is a commented example configuration.
 
 ```toml
 {{#rustdoc_include ../../examples/server_container.toml}}
diff --git a/book/src/sync/concepts.md b/book/src/sync/concepts.md
index 4c96a45b3..69b0831d4 100644
--- a/book/src/sync/concepts.md
+++ b/book/src/sync/concepts.md
@@ -4,7 +4,7 @@
 
 In some environments Kanidm may be the first Identity Management system introduced. However many
 existing environments have existing IDM systems that are well established and in use. To allow
-Kanidm to work with these, it is possible to synchronised data between these IDM systems.
+Kanidm to work with these, it is possible to synchronise data between these IDM systems.
 
 Currently Kanidm can consume (import) data from another IDM system. There are two major use cases
 for this:
@@ -13,7 +13,7 @@ for this:
 - Migrating from an existing IDM to Kanidm
 
 An incoming IDM data source is bound to Kanidm by a sync account. All synchronised entries will have
-a reference to the sync account that they came from defined by their "sync parent uuid". While an
+a reference to the sync account that they came from defined by their `sync_parent_uuid`. While an
 entry is owned by a sync account we refer to the sync account as having authority over the content
 of that entry.
 
diff --git a/book/src/sync/freeipa.md b/book/src/sync/freeipa.md
index 5fb241bfa..541fb0743 100644
--- a/book/src/sync/freeipa.md
+++ b/book/src/sync/freeipa.md
@@ -7,7 +7,8 @@ Kanidm is able to synchronise from FreeIPA for the purposes of coexistence or mi
 
 ## Installing the FreeIPA Sync Tool
 
-See [installing the client tools](../installing_client_tools.md).
+See [installing the client tools](../installing_client_tools.md). The ipa sync tool is part of the
+[tools container](../installing_client_tools.md#tools-container).
 
 ## Configure the FreeIPA Sync Tool
 
@@ -17,9 +18,9 @@ communicate to both sides.
 Like other components of Kanidm, the FreeIPA sync tool will read your /etc/kanidm/config if present
 to understand how to connect to Kanidm.
 
-The sync tool specific components are configured in it's own configuration file.
+The sync tool specific components are configured in its own configuration file.
 
-```rust
+```toml
 {{#rustdoc_include ../../../examples/kanidm-ipa-sync}}
 ```
 
@@ -31,6 +32,7 @@ In addition to this, you must make some configuration changes to FreeIPA to enab
 You can find the name of your 389 Directory Server instance with:
 
 ```bash
+# Run on the FreeIPA server
 dsconf --list
 ```
 
@@ -38,6 +40,7 @@ Using this you can show the current status of the retro changelog plugin to see
 change it's configuration.
 
 ```bash
+# Run on the FreeIPA server
 dsconf <instance name> plugin retro-changelog show
 dsconf slapd-DEV-KANIDM-COM plugin retro-changelog show
 ```
@@ -87,6 +90,18 @@ option "--schedule" on the cli
 
 ```bash
 kanidm-ipa-sync [-c /path/to/kanidm/config] -i /path/to/kanidm-ipa-sync --schedule
+kanidm-ipa-sync -i /etc/kanidm/ipa-sync --schedule
+```
+
+As the sync tool is part of the tools container, you can run this with:
+
+```bash
+docker create --name kanidm-ipa-sync \
+  --user uid:gid \
+  -p 12345:12345 \
+  -v /etc/kanidm/config:/etc/kanidm/config:ro \
+  -v /path/to/ipa-sync:/etc/kanidm/ipa-sync:ro \
+  kanidm-ipa-sync -i /etc/kanidm/ipa-sync --schedule
 ```
 
 ## Monitoring the Sync Tool
@@ -105,4 +120,4 @@ Ok
 ```
 
 It's important to note no details are revealed via the status socket, and is purely for Ok or Err
-status of the last sync.
+status of the last sync. This status socket is suitable for monitoring from tools such as Nagios.
diff --git a/book/src/sync/ldap.md b/book/src/sync/ldap.md
index 217fa44f2..6286093af 100644
--- a/book/src/sync/ldap.md
+++ b/book/src/sync/ldap.md
@@ -20,7 +20,7 @@ understand how to connect to Kanidm.
 
 The sync tool specific components are configured in it's own configuration file.
 
-```rust
+```toml
 {{#rustdoc_include ../../../examples/kanidm-ldap-sync}}
 ```
 
@@ -113,6 +113,18 @@ option "--schedule" on the cli
 
 ```bash
 kanidm-ldap-sync [-c /path/to/kanidm/config] -i /path/to/kanidm-ldap-sync --schedule
+kanidm-ldap-sync -i /etc/kanidm/ldap-sync --schedule
+```
+
+As the sync tool is part of the tools container, you can run this with:
+
+```bash
+docker create --name kanidm-ldap-sync \
+  --user uid:gid \
+  -p 12345:12345 \
+  -v /etc/kanidm/config:/etc/kanidm/config:ro \
+  -v /path/to/ldap-sync:/etc/kanidm/ldap-sync:ro \
+  kanidm-ipa-sync -i /etc/kanidm/ldap-sync --schedule
 ```
 
 ## Monitoring the Sync Tool
@@ -131,4 +143,4 @@ Ok
 ```
 
 It's important to note no details are revealed via the status socket, and is purely for Ok or Err
-status of the last sync.
+status of the last sync. This status socket is suitable for monitoring from tools such as Nagios.
diff --git a/examples/kanidm-ipa-sync b/examples/kanidm-ipa-sync
index 152e8ba7d..53e12cc57 100644
--- a/examples/kanidm-ipa-sync
+++ b/examples/kanidm-ipa-sync
@@ -1,4 +1,3 @@
-
 # The sync account token as generated by "system sync generate-token".
 sync_token = "eyJhb..."
 
diff --git a/examples/server.toml b/examples/server.toml
index 0506c331b..5c9a9a343 100644
--- a/examples/server.toml
+++ b/examples/server.toml
@@ -23,8 +23,8 @@ bindaddress = "[::]:443"
 #   The path to the kanidm database.
 db_path = "/var/lib/private/kanidm/kanidm.db"
 #
-#   If you have a known filesystem, kanidm can tune database
-#   to match. Valid choices are:
+#   If you have a known filesystem, kanidm can tune the 
+#   database page size to match. Valid choices are:
 #   [zfs, other]
 #   If you are unsure about this leave it as the default
 #   (other). After changing this
@@ -40,6 +40,8 @@ db_path = "/var/lib/private/kanidm/kanidm.db"
 #   The number of entries to store in the in-memory cache.
 #   Minimum value is 256. If unset
 #   an automatic heuristic is used to scale this.
+#   You should only adjust this value if you experience
+#   pressure on your system.
 # db_arc_size = 2048
 #
 #   TLS chain and key in pem format. Both must be present
@@ -77,24 +79,6 @@ domain = "idm.example.com"
 #   origin = "https://idm.example.com"
 origin = "https://idm.example.com:8443"
 #
-#   The role of this server. This affects available features
-#   and how replication may interact.
-#   Valid roles are:
-#   - WriteReplica
-#     This server provides all functionality of Kanidm. It
-#     allows authentication, writes, and
-#     the web user interface to be served.
-#   - WriteReplicaNoUI
-#     This server is the same as a WriteReplica, but does NOT
-#     offer the web user interface.
-#   - ReadOnlyReplica
-#     This server will not writes initiated by clients. It
-#     supports authentication and reads,
-#     and must have a replication agreement as a source of
-#     its data.
-#   Defaults to "WriteReplica".
-# role = "WriteReplica"
-#
 [online_backup]
 #   The path to the output folder for online backups
 path = "/var/lib/private/kanidm/backups/"
diff --git a/examples/server_container.toml b/examples/server_container.toml
index 467ef1e2b..0d7a21f8c 100644
--- a/examples/server_container.toml
+++ b/examples/server_container.toml
@@ -23,8 +23,8 @@ bindaddress = "[::]:8443"
 #   The path to the kanidm database.
 db_path = "/data/kanidm.db"
 #
-#   If you have a known filesystem, kanidm can tune database
-#   to match. Valid choices are:
+#   If you have a known filesystem, kanidm can tune the 
+#   database page size to match. Valid choices are:
 #   [zfs, other]
 #   If you are unsure about this leave it as the default
 #   (other). After changing this
@@ -40,6 +40,8 @@ db_path = "/data/kanidm.db"
 #   The number of entries to store in the in-memory cache.
 #   Minimum value is 256. If unset
 #   an automatic heuristic is used to scale this.
+#   You should only adjust this value if you experience
+#   pressure on your system.
 # db_arc_size = 2048
 #
 #   TLS chain and key in pem format. Both must be present
@@ -77,24 +79,6 @@ domain = "idm.example.com"
 #   origin = "https://idm.example.com"
 origin = "https://idm.example.com:8443"
 #
-#   The role of this server. This affects available features
-#   and how replication may interact.
-#   Valid roles are:
-#   - WriteReplica
-#     This server provides all functionality of Kanidm. It
-#     allows authentication, writes, and
-#     the web user interface to be served.
-#   - WriteReplicaNoUI
-#     This server is the same as a WriteReplica, but does NOT
-#     offer the web user interface.
-#   - ReadOnlyReplica
-#     This server will not writes initiated by clients. It
-#     supports authentication and reads,
-#     and must have a replication agreement as a source of
-#     its data.
-#   Defaults to "WriteReplica".
-# role = "WriteReplica"
-#
 [online_backup]
 #   The path to the output folder for online backups
 path = "/data/kanidm/backups/"
diff --git a/libs/crypto/src/lib.rs b/libs/crypto/src/lib.rs
index cc9b0494c..775c65005 100644
--- a/libs/crypto/src/lib.rs
+++ b/libs/crypto/src/lib.rs
@@ -823,7 +823,6 @@ impl Password {
             })
             .map_err(|_| CryptoError::Argon2)
             .map(|material| Password { material })
-            .map_err(|e| e.into())
     }
 
     pub fn new_argon2id_tpm(
@@ -857,7 +856,6 @@ impl Password {
                 key,
             })
             .map(|material| Password { material })
-            .map_err(|e| e.into())
     }
 
     #[inline]
diff --git a/libs/crypto/src/serialise.rs b/libs/crypto/src/serialise.rs
index 3e953555e..faa844512 100644
--- a/libs/crypto/src/serialise.rs
+++ b/libs/crypto/src/serialise.rs
@@ -14,7 +14,7 @@ pub mod pkeyb64 {
             error!(?err, "openssl private_key_to_der");
             S::Error::custom("openssl private_key_to_der")
         })?;
-        let s = general_purpose::URL_SAFE.encode(&der);
+        let s = general_purpose::URL_SAFE_NO_PAD.encode(der);
 
         ser.serialize_str(&s)
     }
@@ -24,10 +24,13 @@ pub mod pkeyb64 {
         D: Deserializer<'de>,
     {
         let raw = <&str>::deserialize(des)?;
-        let s = general_purpose::URL_SAFE.decode(raw).map_err(|err| {
-            error!(?err, "base64 url-safe invalid");
-            D::Error::custom("base64 url-safe invalid")
-        })?;
+        let s = general_purpose::URL_SAFE_NO_PAD
+            .decode(raw)
+            .or_else(|_| general_purpose::URL_SAFE.decode(raw))
+            .map_err(|err| {
+                error!(?err, "base64 url-safe invalid");
+                D::Error::custom("base64 url-safe invalid")
+            })?;
 
         PKey::private_key_from_der(&s).map_err(|err| {
             error!(?err, "openssl pkey invalid der");
@@ -51,7 +54,7 @@ pub mod x509b64 {
                 error!(?err, "openssl cert to_der");
                 err.into()
             })
-            .map(|der| general_purpose::URL_SAFE.encode(&der))
+            .map(|der| general_purpose::URL_SAFE.encode(der))
     }
 
     pub fn serialize<S>(cert: &X509, ser: S) -> Result<S::Ok, S::Error>
@@ -62,7 +65,7 @@ pub mod x509b64 {
             error!(?err, "openssl cert to_der");
             S::Error::custom("openssl private_key_to_der")
         })?;
-        let s = general_purpose::URL_SAFE.encode(&der);
+        let s = general_purpose::URL_SAFE_NO_PAD.encode(der);
 
         ser.serialize_str(&s)
     }
@@ -72,10 +75,13 @@ pub mod x509b64 {
         D: Deserializer<'de>,
     {
         let raw = <&str>::deserialize(des)?;
-        let s = general_purpose::URL_SAFE.decode(raw).map_err(|err| {
-            error!(?err, "base64 url-safe invalid");
-            D::Error::custom("base64 url-safe invalid")
-        })?;
+        let s = general_purpose::URL_SAFE_NO_PAD
+            .decode(raw)
+            .or_else(|_| general_purpose::URL_SAFE.decode(raw))
+            .map_err(|err| {
+                error!(?err, "base64 url-safe invalid");
+                D::Error::custom("base64 url-safe invalid")
+            })?;
 
         X509::from_der(&s).map_err(|err| {
             error!(?err, "openssl x509 invalid der");
diff --git a/server/core/src/actors/v1_write.rs b/server/core/src/actors/v1_write.rs
index ac560e8c7..85faf6444 100644
--- a/server/core/src/actors/v1_write.rs
+++ b/server/core/src/actors/v1_write.rs
@@ -7,7 +7,7 @@ use kanidm_proto::v1::{
     ModifyRequest, OperationError,
 };
 use time::OffsetDateTime;
-use tracing::{info, instrument, span, trace, Level};
+use tracing::{info, instrument, span, trace, Instrument, Level};
 use uuid::Uuid;
 
 use kanidmd_lib::{
@@ -1593,18 +1593,21 @@ impl QueryServerWriteV1 {
 
     pub(crate) async fn handle_delayedaction(&self, da: DelayedAction) {
         let eventid = Uuid::new_v4();
-        let nspan = span!(Level::INFO, "process_delayed_action", uuid = ?eventid);
-        let _span = nspan.enter();
+        let span = span!(Level::INFO, "process_delayed_action", uuid = ?eventid);
 
-        trace!("Begin delayed action ...");
-        let ct = duration_from_epoch_now();
-        let mut idms_prox_write = self.idms.proxy_write(ct).await;
-        if let Err(res) = idms_prox_write
-            .process_delayedaction(da, ct)
-            .and_then(|_| idms_prox_write.commit())
-        {
-            admin_info!(?res, "delayed action error");
+        async {
+            trace!("Begin delayed action ...");
+            let ct = duration_from_epoch_now();
+            let mut idms_prox_write = self.idms.proxy_write(ct).await;
+            if let Err(res) = idms_prox_write
+                .process_delayedaction(da, ct)
+                .and_then(|_| idms_prox_write.commit())
+            {
+                info!(?res, "delayed action error");
+            }
         }
+        .instrument(span)
+        .await
     }
 
     #[instrument(
diff --git a/server/core/src/admin.rs b/server/core/src/admin.rs
index beb1802e7..e768e733d 100644
--- a/server/core/src/admin.rs
+++ b/server/core/src/admin.rs
@@ -22,12 +22,14 @@ pub enum AdminTaskRequest {
     RecoverAccount { name: String },
     ShowReplicationCertificate,
     RenewReplicationCertificate,
+    RefreshReplicationConsumer,
 }
 
 #[derive(Serialize, Deserialize, Debug)]
 pub enum AdminTaskResponse {
     RecoverAccount { password: String },
     ShowReplicationCertificate { cert: String },
+    Success,
     Error,
 }
 
@@ -237,6 +239,35 @@ async fn renew_replication_certificate(ctrl_tx: &mut mpsc::Sender<ReplCtrl>) ->
     }
 }
 
+async fn replication_consumer_refresh(ctrl_tx: &mut mpsc::Sender<ReplCtrl>) -> AdminTaskResponse {
+    let (tx, rx) = oneshot::channel();
+
+    if ctrl_tx
+        .send(ReplCtrl::RefreshConsumer { respond: tx })
+        .await
+        .is_err()
+    {
+        error!("replication control channel has shutdown");
+        return AdminTaskResponse::Error;
+    }
+
+    match rx.await {
+        Ok(mut refresh_rx) => {
+            if let Some(()) = refresh_rx.recv().await {
+                info!("Replication refresh success");
+                AdminTaskResponse::Success
+            } else {
+                error!("Replication refresh failed. Please inspect the logs.");
+                AdminTaskResponse::Error
+            }
+        }
+        Err(_) => {
+            error!("replication control channel did not respond with refresh status.");
+            AdminTaskResponse::Error
+        }
+    }
+}
+
 async fn handle_client(
     sock: UnixStream,
     server: &'static QueryServerWriteV1,
@@ -251,7 +282,6 @@ async fn handle_client(
         // Setup the logging span
         let eventid = Uuid::new_v4();
         let nspan = span!(Level::INFO, "handle_admin_client_request", uuid = ?eventid);
-        // let _span = nspan.enter();
 
         let resp = async {
             match req {
@@ -278,6 +308,13 @@ async fn handle_client(
                         AdminTaskResponse::Error
                     }
                 },
+                AdminTaskRequest::RefreshReplicationConsumer => match repl_ctrl_tx.as_mut() {
+                    Some(ctrl_tx) => replication_consumer_refresh(ctrl_tx).await,
+                    None => {
+                        error!("replication not configured, unable to refresh consumer.");
+                        AdminTaskResponse::Error
+                    }
+                },
             }
         }
         .instrument(nspan)
diff --git a/server/core/src/config.rs b/server/core/src/config.rs
index 8c965ed05..3fb589d2b 100644
--- a/server/core/src/config.rs
+++ b/server/core/src/config.rs
@@ -302,8 +302,8 @@ impl fmt::Display for Configuration {
     }
 }
 
-impl Configuration {
-    pub fn new() -> Self {
+impl Default for Configuration {
+    fn default() -> Self {
         Configuration {
             address: DEFAULT_SERVER_ADDRESS.to_string(),
             ldapaddress: None,
@@ -331,6 +331,12 @@ impl Configuration {
             integration_repl_config: None,
         }
     }
+}
+
+impl Configuration {
+    pub fn new() -> Self {
+        Self::default()
+    }
 
     pub fn new_for_test() -> Self {
         Configuration {
diff --git a/server/core/src/repl/codec.rs b/server/core/src/repl/codec.rs
index c9b68fe89..d0df7f2ee 100644
--- a/server/core/src/repl/codec.rs
+++ b/server/core/src/repl/codec.rs
@@ -1,4 +1,4 @@
-use bytes::{BufMut, BytesMut};
+use bytes::{Buf, BufMut, BytesMut};
 use serde::{de::DeserializeOwned, Deserialize, Serialize};
 use std::io;
 use tokio_util::codec::{Decoder, Encoder};
@@ -76,16 +76,19 @@ impl Encoder<SupplierResponse> for SupplierCodec {
 }
 
 fn encode_length_checked_json<R: Serialize>(msg: R, dst: &mut BytesMut) -> Result<(), io::Error> {
+    // First, if there is anything already in dst, we should split past it.
+    let mut work = dst.split_off(dst.len());
+
     // Null the head of the buffer.
     let zero_len = u64::MIN.to_be_bytes();
-    dst.extend_from_slice(&zero_len);
+    work.extend_from_slice(&zero_len);
 
     // skip the buffer ahead 8 bytes.
     // Remember, this split returns the *already set* bytes.
     // ⚠️  Can't use split or split_at - these return the
     // len bytes into a new bytes mut which confuses unsplit
     // by appending the value when we need to append our json.
-    let json_buf = dst.split_off(zero_len.len());
+    let json_buf = work.split_off(zero_len.len());
 
     let mut json_writer = json_buf.writer();
 
@@ -99,15 +102,17 @@ fn encode_length_checked_json<R: Serialize>(msg: R, dst: &mut BytesMut) -> Resul
     let final_len = json_buf.len() as u64;
     let final_len_bytes = final_len.to_be_bytes();
 
-    if final_len_bytes.len() != dst.len() {
+    if final_len_bytes.len() != work.len() {
         error!("consumer buffer size error");
         return Err(io::Error::new(io::ErrorKind::Other, "buffer length error"));
     }
 
-    dst.copy_from_slice(&final_len_bytes);
+    work.copy_from_slice(&final_len_bytes);
 
     // Now stitch them back together.
-    dst.unsplit(json_buf);
+    work.unsplit(json_buf);
+
+    dst.unsplit(work);
 
     Ok(())
 }
@@ -147,7 +152,7 @@ fn decode_length_checked_json<T: DeserializeOwned>(
         ));
     }
 
-    if (src.len() as u64) < req_len {
+    if (json_bytes.len() as u64) < req_len {
         trace!(
             "Insufficient bytes for json, need: {} have: {}",
             req_len,
@@ -156,6 +161,10 @@ fn decode_length_checked_json<T: DeserializeOwned>(
         return Ok(None);
     }
 
+    // If there are excess bytes, we need to limit our slice to that view.
+    debug_assert!(req_len as usize <= json_bytes.len());
+    let (json_bytes, _remainder) = json_bytes.split_at(req_len as usize);
+
     // Okay, we have enough. Lets go.
     let res = serde_json::from_slice(json_bytes)
         .map(|msg| Some(msg))
@@ -169,8 +178,7 @@ fn decode_length_checked_json<T: DeserializeOwned>(
     if src.len() as u64 == req_len {
         src.clear();
     } else {
-        let mut rem = src.split_off((8 + req_len) as usize);
-        std::mem::swap(&mut rem, src);
+        src.advance((8 + req_len) as usize);
     };
 
     res
@@ -246,5 +254,29 @@ mod tests {
             Ok(Some(SupplierResponse::Pong))
         ));
         assert!(buf.is_empty());
+
+        // Make two requests in a row.
+        buf.clear();
+        let mut supplier_codec = SupplierCodec::new(32);
+
+        assert!(consumer_codec
+            .encode(ConsumerRequest::Ping, &mut buf)
+            .is_ok());
+        assert!(consumer_codec
+            .encode(ConsumerRequest::Ping, &mut buf)
+            .is_ok());
+
+        assert!(matches!(
+            supplier_codec.decode(&mut buf),
+            Ok(Some(ConsumerRequest::Ping))
+        ));
+        assert!(!buf.is_empty());
+        assert!(matches!(
+            supplier_codec.decode(&mut buf),
+            Ok(Some(ConsumerRequest::Ping))
+        ));
+
+        // The buf will have been cleared by the supplier codec here.
+        assert!(buf.is_empty());
     }
 }
diff --git a/server/core/src/repl/mod.rs b/server/core/src/repl/mod.rs
index 683498a4c..aa9257019 100644
--- a/server/core/src/repl/mod.rs
+++ b/server/core/src/repl/mod.rs
@@ -12,9 +12,10 @@ use tokio::net::{TcpListener, TcpStream};
 use tokio::sync::broadcast;
 use tokio::sync::mpsc;
 use tokio::sync::oneshot;
+use tokio::sync::Mutex;
 use tokio::time::{interval, sleep, timeout};
 use tokio_openssl::SslStream;
-use tokio_util::codec::{FramedRead, FramedWrite};
+use tokio_util::codec::{Framed, FramedRead, FramedWrite};
 use tracing::{error, Instrument};
 use url::Url;
 use uuid::Uuid;
@@ -36,8 +37,21 @@ use self::codec::{ConsumerRequest, SupplierResponse};
 mod codec;
 
 pub(crate) enum ReplCtrl {
-    GetCertificate { respond: oneshot::Sender<X509> },
-    RenewCertificate { respond: oneshot::Sender<bool> },
+    GetCertificate {
+        respond: oneshot::Sender<X509>,
+    },
+    RenewCertificate {
+        respond: oneshot::Sender<bool>,
+    },
+    RefreshConsumer {
+        respond: oneshot::Sender<mpsc::Receiver<()>>,
+    },
+}
+
+#[derive(Debug, Clone)]
+enum ReplConsumerCtrl {
+    Stop,
+    Refresh(Arc<Mutex<(bool, mpsc::Sender<()>)>>),
 }
 
 pub(crate) async fn create_repl_server(
@@ -76,21 +90,21 @@ pub(crate) async fn create_repl_server(
 }
 
 #[instrument(level = "info", skip_all)]
-async fn repl_run_consumer(
-    max_frame_bytes: usize,
+async fn repl_consumer_connect_supplier(
     domain: &str,
     sock_addrs: &[SocketAddr],
     tls_connector: &SslConnector,
-    automatic_refresh: bool,
-    idms: &IdmServer,
-) {
-    let replica_connect_timeout = Duration::from_secs(2);
-
+    consumer_conn_settings: &ConsumerConnSettings,
+) -> Option<Framed<SslStream<TcpStream>, codec::ConsumerCodec>> {
     // This is pretty gnarly, but we need to loop to try out each socket addr.
     for sock_addr in sock_addrs {
         debug!("Connecting to {} replica via {}", domain, sock_addr);
 
-        let tcpstream = match timeout(replica_connect_timeout, TcpStream::connect(sock_addr)).await
+        let tcpstream = match timeout(
+            consumer_conn_settings.replica_connect_timeout,
+            TcpStream::connect(sock_addr),
+        )
+        .await
         {
             Ok(Ok(tc)) => tc,
             Ok(Err(err)) => {
@@ -119,112 +133,77 @@ async fn repl_run_consumer(
             error!("replication client TLS accept error, continuing -> {:?}", e);
             continue;
         };
-        let (r, w) = tokio::io::split(tlsstream);
-        let mut r = FramedRead::new(r, codec::ConsumerCodec::new(max_frame_bytes));
-        let mut w = FramedWrite::new(w, codec::ConsumerCodec::new(max_frame_bytes));
 
-        // Perform incremental.
-        let consumer_ruv_range = {
-            let mut read_txn = idms.proxy_read().await;
-            match read_txn.qs_read.consumer_get_state() {
-                Ok(ruv_range) => ruv_range,
-                Err(err) => {
-                    error!(
-                        ?err,
-                        "consumer ruv range could not be accessed, unable to continue."
-                    );
-                    break;
-                }
-            }
-        };
+        let supplier_conn = Framed::new(
+            tlsstream,
+            codec::ConsumerCodec::new(consumer_conn_settings.max_frame_bytes),
+        );
 
-        if let Err(err) = w
-            .send(ConsumerRequest::Incremental(consumer_ruv_range))
+        return Some(supplier_conn);
+    }
+
+    error!("Unable to connect to supplier.");
+    None
+}
+
+async fn repl_run_consumer_refresh(
+    refresh_coord: Arc<Mutex<(bool, mpsc::Sender<()>)>>,
+    domain: &str,
+    sock_addrs: &[SocketAddr],
+    tls_connector: &SslConnector,
+    idms: &IdmServer,
+    consumer_conn_settings: &ConsumerConnSettings,
+) {
+    // Take the refresh lock. Note that every replication consumer *should* end up here
+    // behind this lock, but only one can proceed. This is what we want!
+
+    let mut refresh_coord_guard = refresh_coord.lock().await;
+
+    // Simple case - task is already done.
+    if refresh_coord_guard.0 {
+        trace!("refresh already completed by another task, return.");
+        return;
+    }
+
+    // okay, we need to proceed.
+    let Some(mut supplier_conn) =
+        repl_consumer_connect_supplier(domain, sock_addrs, tls_connector, consumer_conn_settings)
             .await
-        {
-            error!(?err, "consumer encode error, unable to continue.");
-            break;
-        }
+    else {
+        return;
+    };
 
-        let changes = if let Some(codec_msg) = r.next().await {
-            match codec_msg {
-                Ok(SupplierResponse::Incremental(changes)) => {
-                    // Success - return to bypass the error message.
-                    changes
-                }
-                Ok(SupplierResponse::Pong) | Ok(SupplierResponse::Refresh(_)) => {
-                    error!("Supplier Response contains invalid State");
-                    break;
-                }
-                Err(err) => {
-                    error!(?err, "consumer decode error, unable to continue.");
-                    break;
-                }
+    // If we fail at anypoint, just RETURN because this leaves the next task to attempt, or
+    // the channel drops and that tells the caller this failed.
+
+    if let Err(err) = supplier_conn.send(ConsumerRequest::Refresh).await {
+        error!(?err, "consumer encode error, unable to continue.");
+        return;
+    }
+
+    let refresh = if let Some(codec_msg) = supplier_conn.next().await {
+        match codec_msg {
+            Ok(SupplierResponse::Refresh(changes)) => {
+                // Success - return to bypass the error message.
+                changes
             }
-        } else {
-            error!("Connection closed");
-            break;
-        };
-
-        // Now apply the changes if possible
-        let consumer_state = {
-            let ct = duration_from_epoch_now();
-            let mut write_txn = idms.proxy_write(ct).await;
-            match write_txn
-                .qs_write
-                .consumer_apply_changes(&changes)
-                .and_then(|cs| write_txn.commit().map(|()| cs))
-            {
-                Ok(state) => state,
-                Err(err) => {
-                    error!(?err, "consumer was not able to apply changes.");
-                    break;
-                }
-            }
-        };
-
-        match consumer_state {
-            ConsumerState::Ok => {
-                info!("Incremental Replication Success");
-                // return to bypass the failure message.
+            Ok(SupplierResponse::Pong) | Ok(SupplierResponse::Incremental(_)) => {
+                error!("Supplier Response contains invalid State");
                 return;
             }
-            ConsumerState::RefreshRequired => {
-                if automatic_refresh {
-                    warn!("Consumer is out of date and must be refreshed. This will happen *now*.");
-                } else {
-                    error!("Consumer is out of date and must be refreshed. You must manually resolve this situation.");
-                    return;
-                };
+            Err(err) => {
+                error!(?err, "consumer decode error, unable to continue.");
+                return;
             }
         }
+    } else {
+        error!("Connection closed");
+        return;
+    };
 
-        if let Err(err) = w.send(ConsumerRequest::Refresh).await {
-            error!(?err, "consumer encode error, unable to continue.");
-            break;
-        }
-
-        let refresh = if let Some(codec_msg) = r.next().await {
-            match codec_msg {
-                Ok(SupplierResponse::Refresh(changes)) => {
-                    // Success - return to bypass the error message.
-                    changes
-                }
-                Ok(SupplierResponse::Pong) | Ok(SupplierResponse::Incremental(_)) => {
-                    error!("Supplier Response contains invalid State");
-                    break;
-                }
-                Err(err) => {
-                    error!(?err, "consumer decode error, unable to continue.");
-                    break;
-                }
-            }
-        } else {
-            error!("Connection closed");
-            break;
-        };
-
-        // Now apply the refresh if possible
+    // Now apply the refresh if possible
+    {
+        // Scope the transaction.
         let ct = duration_from_epoch_now();
         let mut write_txn = idms.proxy_write(ct).await;
         if let Err(err) = write_txn
@@ -233,14 +212,157 @@ async fn repl_run_consumer(
             .and_then(|cs| write_txn.commit().map(|()| cs))
         {
             error!(?err, "consumer was not able to apply refresh.");
-            break;
+            return;
         }
+    }
 
-        warn!("Replication refresh was successful.");
+    // Now mark the refresh as complete AND indicate it to the channel.
+    refresh_coord_guard.0 = true;
+    if refresh_coord_guard.1.send(()).await.is_err() {
+        warn!("Unable to signal to caller that refresh has completed.");
+    }
+
+    // Here the coord guard will drop and every other task proceeds.
+
+    warn!("Replication refresh was successful.");
+}
+
+async fn repl_run_consumer(
+    domain: &str,
+    sock_addrs: &[SocketAddr],
+    tls_connector: &SslConnector,
+    automatic_refresh: bool,
+    idms: &IdmServer,
+    consumer_conn_settings: &ConsumerConnSettings,
+) {
+    let Some(mut supplier_conn) =
+        repl_consumer_connect_supplier(domain, sock_addrs, tls_connector, consumer_conn_settings)
+            .await
+    else {
+        return;
+    };
+
+    // Perform incremental.
+    let consumer_ruv_range = {
+        let mut read_txn = idms.proxy_read().await;
+        match read_txn.qs_read.consumer_get_state() {
+            Ok(ruv_range) => ruv_range,
+            Err(err) => {
+                error!(
+                    ?err,
+                    "consumer ruv range could not be accessed, unable to continue."
+                );
+                return;
+            }
+        }
+    };
+
+    if let Err(err) = supplier_conn
+        .send(ConsumerRequest::Incremental(consumer_ruv_range))
+        .await
+    {
+        error!(?err, "consumer encode error, unable to continue.");
         return;
     }
 
-    error!("Unable to complete replication successfully.");
+    let changes = if let Some(codec_msg) = supplier_conn.next().await {
+        match codec_msg {
+            Ok(SupplierResponse::Incremental(changes)) => {
+                // Success - return to bypass the error message.
+                changes
+            }
+            Ok(SupplierResponse::Pong) | Ok(SupplierResponse::Refresh(_)) => {
+                error!("Supplier Response contains invalid State");
+                return;
+            }
+            Err(err) => {
+                error!(?err, "consumer decode error, unable to continue.");
+                return;
+            }
+        }
+    } else {
+        error!("Connection closed");
+        return;
+    };
+
+    // Now apply the changes if possible
+    let consumer_state = {
+        let ct = duration_from_epoch_now();
+        let mut write_txn = idms.proxy_write(ct).await;
+        match write_txn
+            .qs_write
+            .consumer_apply_changes(&changes)
+            .and_then(|cs| write_txn.commit().map(|()| cs))
+        {
+            Ok(state) => state,
+            Err(err) => {
+                error!(?err, "consumer was not able to apply changes.");
+                return;
+            }
+        }
+    };
+
+    match consumer_state {
+        ConsumerState::Ok => {
+            info!("Incremental Replication Success");
+            // return to bypass the failure message.
+            return;
+        }
+        ConsumerState::RefreshRequired => {
+            if automatic_refresh {
+                warn!("Consumer is out of date and must be refreshed. This will happen *now*.");
+            } else {
+                error!("Consumer is out of date and must be refreshed. You must manually resolve this situation.");
+                return;
+            };
+        }
+    }
+
+    if let Err(err) = supplier_conn.send(ConsumerRequest::Refresh).await {
+        error!(?err, "consumer encode error, unable to continue.");
+        return;
+    }
+
+    let refresh = if let Some(codec_msg) = supplier_conn.next().await {
+        match codec_msg {
+            Ok(SupplierResponse::Refresh(changes)) => {
+                // Success - return to bypass the error message.
+                changes
+            }
+            Ok(SupplierResponse::Pong) | Ok(SupplierResponse::Incremental(_)) => {
+                error!("Supplier Response contains invalid State");
+                return;
+            }
+            Err(err) => {
+                error!(?err, "consumer decode error, unable to continue.");
+                return;
+            }
+        }
+    } else {
+        error!("Connection closed");
+        return;
+    };
+
+    // Now apply the refresh if possible
+    let ct = duration_from_epoch_now();
+    let mut write_txn = idms.proxy_write(ct).await;
+    if let Err(err) = write_txn
+        .qs_write
+        .consumer_apply_refresh(&refresh)
+        .and_then(|cs| write_txn.commit().map(|()| cs))
+    {
+        error!(?err, "consumer was not able to apply refresh.");
+        return;
+    }
+
+    warn!("Replication refresh was successful.");
+}
+
+#[derive(Debug, Clone)]
+struct ConsumerConnSettings {
+    max_frame_bytes: usize,
+    task_poll_interval: Duration,
+    replica_connect_timeout: Duration,
 }
 
 async fn repl_task(
@@ -248,9 +370,8 @@ async fn repl_task(
     client_key: PKey<Private>,
     client_cert: X509,
     supplier_cert: X509,
-    max_frame_bytes: usize,
-    task_poll_interval: Duration,
-    mut task_rx: broadcast::Receiver<()>,
+    consumer_conn_settings: ConsumerConnSettings,
+    mut task_rx: broadcast::Receiver<ReplConsumerCtrl>,
     automatic_refresh: bool,
     idms: Arc<IdmServer>,
 ) {
@@ -325,29 +446,48 @@ async fn repl_task(
     ssl_builder.set_verify(SslVerifyMode::PEER);
     let tls_connector = ssl_builder.build();
 
-    let mut repl_interval = interval(task_poll_interval);
+    let mut repl_interval = interval(consumer_conn_settings.task_poll_interval);
 
     info!("Replica task for {} has started.", origin);
 
     // Okay, all the parameters are setup. Now we wait on our interval.
     loop {
         tokio::select! {
-            Ok(()) = task_rx.recv() => {
-                break;
+            Ok(task) = task_rx.recv() => {
+                match task {
+                    ReplConsumerCtrl::Stop => break,
+                    ReplConsumerCtrl::Refresh ( refresh_coord ) => {
+                        let eventid = Uuid::new_v4();
+                        let span = info_span!("replication_run_consumer_refresh", uuid = ?eventid);
+                        // let _enter = span.enter();
+                        repl_run_consumer_refresh(
+                            refresh_coord,
+                            domain,
+                            &socket_addrs,
+                            &tls_connector,
+                            &idms,
+                            &consumer_conn_settings
+                        )
+                        .instrument(span)
+                        .await
+                    }
+                }
             }
             _ = repl_interval.tick() => {
                 // Interval passed, attempt a replication run.
                 let eventid = Uuid::new_v4();
                 let span = info_span!("replication_run_consumer", uuid = ?eventid);
-                let _enter = span.enter();
+                // let _enter = span.enter();
                 repl_run_consumer(
-                    max_frame_bytes,
                     domain,
                     &socket_addrs,
                     &tls_connector,
                     automatic_refresh,
-                    &idms
-                ).await;
+                    &idms,
+                    &consumer_conn_settings
+                )
+                .instrument(span)
+                .await;
             }
         }
     }
@@ -447,9 +587,16 @@ async fn repl_acceptor(
     // Persistent parts
     // These all probably need changes later ...
     let task_poll_interval = Duration::from_secs(10);
+    let replica_connect_timeout = Duration::from_secs(2);
     let retry_timeout = Duration::from_secs(60);
     let max_frame_bytes = 268435456;
 
+    let consumer_conn_settings = ConsumerConnSettings {
+        max_frame_bytes,
+        task_poll_interval,
+        replica_connect_timeout,
+    };
+
     // Setup a broadcast to control our tasks.
     let (task_tx, task_rx1) = broadcast::channel(2);
     // Note, we drop this task here since each task will re-subscribe. That way the
@@ -464,6 +611,13 @@ async fn repl_acceptor(
     // In future we need to update this from the KRC if configured, and we default this
     // to "empty". But if this map exists in the config, we have to always use that.
     let replication_node_map = repl_config.manual.clone();
+    let domain_name = match repl_config.origin.domain() {
+        Some(n) => n.to_string(),
+        None => {
+            error!("Unable to start replication, replication origin does not contain a valid domain name.");
+            return;
+        }
+    };
 
     // This needs to have an event loop that can respond to changes.
     // For now we just design it to reload ssl if the map changes internally.
@@ -474,7 +628,7 @@ async fn repl_acceptor(
         // no tasks currently listening on the channel.
         info!("Stopping {} Replication Tasks ...", task_handles.len());
         debug_assert!(task_handles.len() >= task_tx.receiver_count());
-        let _ = task_tx.send(());
+        let _ = task_tx.send(ReplConsumerCtrl::Stop);
         for task_handle in task_handles.drain(..) {
             // Let each task join.
             let res: Result<(), _> = task_handle.await;
@@ -494,7 +648,7 @@ async fn repl_acceptor(
             let mut idms_prox_write = idms.proxy_write(ct).await;
             idms_prox_write
                 .qs_write
-                .supplier_get_key_cert()
+                .supplier_get_key_cert(&domain_name)
                 .and_then(|res| idms_prox_write.commit().map(|()| res))
         };
 
@@ -548,8 +702,7 @@ async fn repl_acceptor(
                         server_key.clone(),
                         server_cert.clone(),
                         supplier_cert.clone(),
-                        max_frame_bytes,
-                        task_poll_interval,
+                        consumer_conn_settings.clone(),
                         task_rx,
                         *automatic_refresh,
                         idms.clone(),
@@ -626,7 +779,7 @@ async fn repl_acceptor(
                             respond
                         } => {
                             let _span = debug_span!("supplier_accept_loop", uuid = ?eventid).entered();
-                            if let Err(_) = respond.send(server_cert.clone()) {
+                            if respond.send(server_cert.clone()).is_err() {
                                 warn!("Server certificate was requested, but requsetor disconnected");
                             } else {
                                 trace!("Sent server certificate via control channel");
@@ -644,7 +797,7 @@ async fn repl_acceptor(
                                     let mut idms_prox_write = idms.proxy_write(ct).await;
                                     idms_prox_write
                                         .qs_write
-                                        .supplier_renew_key_cert()
+                                        .supplier_renew_key_cert(&domain_name)
                                         .and_then(|res| idms_prox_write.commit().map(|()| res))
                                 };
 
@@ -654,7 +807,7 @@ async fn repl_acceptor(
                                     error!(?err, "failed to renew server certificate");
                                 }
 
-                                if let Err(_) = respond.send(success) {
+                                if respond.send(success).is_err() {
                                     warn!("Server certificate renewal was requested, but requsetor disconnected");
                                 } else {
                                     trace!("Sent server certificate renewal status via control channel");
@@ -666,6 +819,30 @@ async fn repl_acceptor(
                             // Start a reload.
                             continue 'event;
                         }
+                        ReplCtrl::RefreshConsumer {
+                            respond
+                        } => {
+                            // Indicate to consumer tasks that they should do a refresh.
+                            let (tx, rx) = mpsc::channel(1);
+
+                            let refresh_coord = Arc::new(
+                                Mutex::new(
+                                (
+                                    false, tx
+                                )
+                                )
+                            );
+
+                            if task_tx.send(ReplConsumerCtrl::Refresh(refresh_coord)).is_err() {
+                                error!("Unable to begin replication consumer refresh, tasks are unable to be notified.");
+                            }
+
+                            if respond.send(rx).is_err() {
+                                warn!("Replication consumer refresh was requested, but requester disconnected");
+                            } else {
+                                trace!("Sent refresh comms channel to requester");
+                            }
+                        }
                     }
                 }
                 // Handle accepts.
@@ -683,7 +860,7 @@ async fn repl_acceptor(
                             let clone_tls_acceptor = tls_acceptor.clone();
                             // We don't care about the join handle here - once a client connects
                             // it sticks to whatever ssl settings it had at launch.
-                            let _ = tokio::spawn(
+                            tokio::spawn(
                                 handle_repl_conn(max_frame_bytes, tcpstream, client_socket_addr, clone_tls_acceptor, clone_idms)
                             );
                         }
@@ -699,7 +876,7 @@ async fn repl_acceptor(
     // Shutdown child tasks.
     info!("Stopping {} Replication Tasks ...", task_handles.len());
     debug_assert!(task_handles.len() >= task_tx.receiver_count());
-    let _ = task_tx.send(());
+    let _ = task_tx.send(ReplConsumerCtrl::Stop);
     for task_handle in task_handles.drain(..) {
         // Let each task join.
         let res: Result<(), _> = task_handle.await;
diff --git a/server/daemon/src/main.rs b/server/daemon/src/main.rs
index d65df5922..c684b7596 100644
--- a/server/daemon/src/main.rs
+++ b/server/daemon/src/main.rs
@@ -78,7 +78,8 @@ impl KanidmdOpt {
                 commands: DbScanOpt::RestoreQuarantined { commonopts, .. },
             }
             | KanidmdOpt::ShowReplicationCertificate { commonopts }
-            | KanidmdOpt::RenewReplicationCertificate { commonopts } => commonopts,
+            | KanidmdOpt::RenewReplicationCertificate { commonopts }
+            | KanidmdOpt::RefreshReplicationConsumer { commonopts, .. } => commonopts,
             KanidmdOpt::RecoverAccount { commonopts, .. } => commonopts,
             KanidmdOpt::DbScan {
                 commands: DbScanOpt::ListIndex(dopt),
@@ -155,7 +156,26 @@ async fn submit_admin_req(path: &str, req: AdminTaskRequest, output_mode: Consol
                 info!(certificate = ?cert)
             }
         },
-        _ => {
+        Some(Ok(AdminTaskResponse::Success)) => match output_mode {
+            ConsoleOutputMode::JSON => {
+                eprintln!("\"success\"")
+            }
+            ConsoleOutputMode::Text => {
+                info!("success")
+            }
+        },
+        Some(Ok(AdminTaskResponse::Error)) => match output_mode {
+            ConsoleOutputMode::JSON => {
+                eprintln!("\"error\"")
+            }
+            ConsoleOutputMode::Text => {
+                info!("Error - you should inspect the logs.")
+            }
+        },
+        Some(Err(err)) => {
+            error!(?err, "Error during admin task operation");
+        }
+        None => {
             error!("Error making request to admin socket");
         }
     }
@@ -365,7 +385,11 @@ async fn main() -> ExitCode {
 
             match &opt.commands  {
                 // we aren't going to touch the DB so we can carry on
-                KanidmdOpt::HealthCheck(_) => (),
+                KanidmdOpt::ShowReplicationCertificate { .. }
+                | KanidmdOpt::RenewReplicationCertificate { .. }
+                | KanidmdOpt::RefreshReplicationConsumer { .. }
+                | KanidmdOpt::RecoverAccount { .. }
+                | KanidmdOpt::HealthCheck(_) => (),
                 _ => {
                     // Okay - Lets now create our lock and go.
                     let klock_path = format!("{}.klock" ,sconfig.db_path.as_str());
@@ -573,6 +597,22 @@ async fn main() -> ExitCode {
                         output_mode,
                     ).await;
                 }
+                KanidmdOpt::RefreshReplicationConsumer {
+                    commonopts,
+                    proceed
+                } => {
+                    info!("Running refresh replication consumer ...");
+                    if !proceed {
+                        error!("Unwilling to proceed. Check --help.");
+
+                    } else {
+                        let output_mode: ConsoleOutputMode = commonopts.output_mode.to_owned().into();
+                        submit_admin_req(config.adminbindpath.as_str(),
+                            AdminTaskRequest::RefreshReplicationConsumer,
+                            output_mode,
+                        ).await;
+                    }
+                }
                 KanidmdOpt::RecoverAccount {
                     name, commonopts
                 } => {
diff --git a/server/daemon/src/opt.rs b/server/daemon/src/opt.rs
index 723fabac6..30bd5cd76 100644
--- a/server/daemon/src/opt.rs
+++ b/server/daemon/src/opt.rs
@@ -164,6 +164,15 @@ enum KanidmdOpt {
         #[clap(flatten)]
         commonopts: CommonOpt,
     },
+    /// Refresh this servers database content with the content from a supplier. This means
+    /// that all local content will be deleted and replaced with the supplier content.
+    RefreshReplicationConsumer {
+        #[clap(flatten)]
+        commonopts: CommonOpt,
+        /// Acknowledge that this database content will be refreshed from a supplier.
+        #[clap(long = "i-want-to-refresh-this-servers-database")]
+        proceed: bool,
+    },
     // #[clap(name = "reset_server_id")]
     // ResetServerId(CommonOpt),
     #[clap(name = "db-scan")]
diff --git a/server/lib/src/be/dbentry.rs b/server/lib/src/be/dbentry.rs
index 3a9606e0e..60d4db314 100644
--- a/server/lib/src/be/dbentry.rs
+++ b/server/lib/src/be/dbentry.rs
@@ -6,8 +6,9 @@ use serde::{Deserialize, Serialize};
 use smartstring::alias::String as AttrString;
 use uuid::Uuid;
 
+use super::dbrepl::{DbEntryChangeState, DbReplMeta};
+use super::dbvalue::{DbValueEmailAddressV1, DbValuePhoneNumberV1, DbValueSetV2, DbValueV1};
 use super::keystorage::{KeyHandle, KeyHandleId};
-use crate::be::dbvalue::{DbValueEmailAddressV1, DbValuePhoneNumberV1, DbValueSetV2, DbValueV1};
 use crate::prelude::entries::Attribute;
 use crate::prelude::OperationError;
 
@@ -28,6 +29,10 @@ pub struct DbEntryV2 {
 pub enum DbEntryVers {
     V1(DbEntryV1),
     V2(DbEntryV2),
+    V3 {
+        changestate: DbEntryChangeState,
+        attrs: BTreeMap<AttrString, DbValueSetV2>,
+    },
 }
 
 #[derive(Serialize, Deserialize, Debug)]
@@ -65,6 +70,14 @@ pub enum DbBackup {
         keyhandles: BTreeMap<KeyHandleId, KeyHandle>,
         entries: Vec<DbEntry>,
     },
+    V4 {
+        db_s_uuid: Uuid,
+        db_d_uuid: Uuid,
+        db_ts_max: Duration,
+        keyhandles: BTreeMap<KeyHandleId, KeyHandle>,
+        repl_meta: DbReplMeta,
+        entries: Vec<DbEntry>,
+    },
 }
 
 fn from_vec_dbval1(attr_val: NonEmpty<DbValueV1>) -> Result<DbValueSetV2, OperationError> {
@@ -438,6 +451,26 @@ impl std::fmt::Debug for DbEntry {
                 }
                 write!(f, "\n        }}")
             }
+            DbEntryVers::V3 { changestate, attrs } => {
+                write!(f, "v3 - {{ ")?;
+                match changestate {
+                    DbEntryChangeState::V1Live { at, changes } => {
+                        writeln!(f, "\nlive {at:>32}")?;
+                        for (attr, cid) in changes {
+                            write!(f, "\n{attr:>32} - {cid} ")?;
+                            if let Some(vs) = attrs.get(attr.as_str()) {
+                                write!(f, "{vs:?}")?;
+                            } else {
+                                write!(f, "-")?;
+                            }
+                        }
+                    }
+                    DbEntryChangeState::V1Tombstone { at } => {
+                        writeln!(f, "\ntombstone {at:>32?}")?;
+                    }
+                }
+                write!(f, "\n        }}")
+            }
         }
     }
 }
@@ -491,6 +524,34 @@ impl std::fmt::Display for DbEntry {
                 }
                 write!(f, "}}")
             }
+            DbEntryVers::V3 { changestate, attrs } => {
+                write!(f, "v3 - {{ ")?;
+                match attrs.get(Attribute::Uuid.as_ref()) {
+                    Some(uuids) => {
+                        write!(f, "{uuids:?}, ")?;
+                    }
+                    None => write!(f, "Uuid(INVALID), ")?,
+                };
+
+                match changestate {
+                    DbEntryChangeState::V1Live { at, changes: _ } => {
+                        write!(f, "created: {at}, ")?;
+                        if let Some(names) = attrs.get(Attribute::Name.as_ref()) {
+                            write!(f, "{names:?}, ")?;
+                        }
+                        if let Some(names) = attrs.get(Attribute::AttributeName.as_ref()) {
+                            write!(f, "{names:?}, ")?;
+                        }
+                        if let Some(names) = attrs.get(Attribute::ClassName.as_ref()) {
+                            write!(f, "{names:?}, ")?;
+                        }
+                    }
+                    DbEntryChangeState::V1Tombstone { at } => {
+                        write!(f, "tombstoned: {at}, ")?;
+                    }
+                }
+                write!(f, "}}")
+            }
         }
     }
 }
diff --git a/server/lib/src/be/dbrepl.rs b/server/lib/src/be/dbrepl.rs
new file mode 100644
index 000000000..aa086bb2d
--- /dev/null
+++ b/server/lib/src/be/dbrepl.rs
@@ -0,0 +1,20 @@
+use super::dbvalue::DbCidV1;
+use std::collections::{BTreeMap, BTreeSet};
+
+use serde::{Deserialize, Serialize};
+
+#[derive(Serialize, Deserialize, Debug)]
+pub enum DbEntryChangeState {
+    V1Live {
+        at: DbCidV1,
+        changes: BTreeMap<String, DbCidV1>,
+    },
+    V1Tombstone {
+        at: DbCidV1,
+    },
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+pub enum DbReplMeta {
+    V1 { ruv: BTreeSet<DbCidV1> },
+}
diff --git a/server/lib/src/be/dbvalue.rs b/server/lib/src/be/dbvalue.rs
index bf6efc4c6..f0cc3f4e1 100644
--- a/server/lib/src/be/dbvalue.rs
+++ b/server/lib/src/be/dbvalue.rs
@@ -16,12 +16,36 @@ use webauthn_rs_core::proto::{COSEKey, UserVerificationPolicy};
 use crate::repl::cid::Cid;
 pub use kanidm_lib_crypto::DbPasswordV1;
 
-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Serialize, Deserialize, Debug, Ord, PartialOrd, PartialEq, Eq)]
 pub struct DbCidV1 {
-    #[serde(rename = "s")]
-    pub server_id: Uuid,
     #[serde(rename = "t")]
     pub timestamp: Duration,
+    #[serde(rename = "s")]
+    pub server_id: Uuid,
+}
+
+impl From<Cid> for DbCidV1 {
+    fn from(Cid { s_uuid, ts }: Cid) -> Self {
+        DbCidV1 {
+            timestamp: ts,
+            server_id: s_uuid,
+        }
+    }
+}
+
+impl From<&Cid> for DbCidV1 {
+    fn from(&Cid { s_uuid, ts }: &Cid) -> Self {
+        DbCidV1 {
+            timestamp: ts,
+            server_id: s_uuid,
+        }
+    }
+}
+
+impl fmt::Display for DbCidV1 {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{:032}-{}", self.timestamp.as_nanos(), self.server_id)
+    }
 }
 
 #[derive(Serialize, Deserialize, Debug)]
diff --git a/server/lib/src/be/idl_arc_sqlite.rs b/server/lib/src/be/idl_arc_sqlite.rs
index 58d677344..5a2870665 100644
--- a/server/lib/src/be/idl_arc_sqlite.rs
+++ b/server/lib/src/be/idl_arc_sqlite.rs
@@ -687,6 +687,18 @@ impl<'a> IdlArcSqliteWriteTransaction<'a> {
         Ok(())
     }
 
+    pub fn get_db_ruv(&self) -> Result<BTreeSet<Cid>, OperationError> {
+        self.db.get_db_ruv()
+    }
+
+    pub fn write_db_ruv<I, J>(&mut self, added: I, removed: J) -> Result<(), OperationError>
+    where
+        I: Iterator<Item = Cid>,
+        J: Iterator<Item = Cid>,
+    {
+        self.db.write_db_ruv(added, removed)
+    }
+
     pub fn get_id2entry_max_id(&self) -> Result<u64, OperationError> {
         Ok(*self.maxid)
     }
diff --git a/server/lib/src/be/idl_sqlite.rs b/server/lib/src/be/idl_sqlite.rs
index 21f98a3cf..dcbbff208 100644
--- a/server/lib/src/be/idl_sqlite.rs
+++ b/server/lib/src/be/idl_sqlite.rs
@@ -1,5 +1,4 @@
-use std::collections::BTreeMap;
-use std::collections::VecDeque;
+use std::collections::{BTreeMap, BTreeSet, VecDeque};
 use std::convert::{TryFrom, TryInto};
 use std::sync::Arc;
 use std::sync::Mutex;
@@ -16,6 +15,7 @@ use rusqlite::{Connection, OpenFlags, OptionalExtension};
 use uuid::Uuid;
 
 use crate::be::dbentry::{DbEntry, DbIdentSpn};
+use crate::be::dbvalue::DbCidV1;
 use crate::be::{BackendConfig, IdList, IdRawEntry, IdxKey, IdxSlope};
 use crate::entry::{Entry, EntryCommitted, EntrySealed};
 use crate::prelude::*;
@@ -1011,7 +1011,7 @@ impl IdlSqliteWriteTransaction {
             .map_err(sqlite_error)
     }
 
-    pub fn migrate_dbentryv1_to_dbentryv2(&self) -> Result<(), OperationError> {
+    fn migrate_dbentryv1_to_dbentryv2(&self) -> Result<(), OperationError> {
         let allids = self.get_identry_raw(&IdList::AllIds)?;
         let raw_entries: Result<Vec<IdRawEntry>, _> = allids
             .into_iter()
@@ -1036,6 +1036,18 @@ impl IdlSqliteWriteTransaction {
         self.write_identries_raw(raw_entries?.into_iter())
     }
 
+    fn migrate_dbentryv2_to_dbentryv3(&self) -> Result<(), OperationError> {
+        // To perform this migration we have to load everything to a valid entry, then
+        // write them all back down once their change states are created.
+        let all_entries = self.get_identry(&IdList::AllIds)?;
+
+        for entry in all_entries {
+            self.write_identry(&entry)?;
+        }
+
+        Ok(())
+    }
+
     pub fn write_uuid2spn(&self, uuid: Uuid, k: Option<&Value>) -> Result<(), OperationError> {
         let uuids = uuid.as_hyphenated().to_string();
         match k {
@@ -1118,6 +1130,90 @@ impl IdlSqliteWriteTransaction {
             .map_err(sqlite_error)
     }
 
+    pub(crate) fn create_db_ruv(&self) -> Result<(), OperationError> {
+        self.get_conn()?
+            .execute(
+                &format!(
+                    "CREATE TABLE IF NOT EXISTS {}.ruv (cid TEXT PRIMARY KEY)",
+                    self.get_db_name()
+                ),
+                [],
+            )
+            .map(|_| ())
+            .map_err(sqlite_error)
+    }
+
+    pub fn get_db_ruv(&self) -> Result<BTreeSet<Cid>, OperationError> {
+        let mut stmt = self
+            .get_conn()?
+            .prepare(&format!("SELECT cid FROM {}.ruv", self.get_db_name()))
+            .map_err(sqlite_error)?;
+
+        let kh_iter = stmt
+            .query_map([], |row| Ok(row.get(0)?))
+            .map_err(sqlite_error)?;
+
+        kh_iter
+            .map(|v| {
+                let ser_cid: String = v.map_err(sqlite_error)?;
+                let db_cid: DbCidV1 = serde_json::from_str(&ser_cid).map_err(serde_json_error)?;
+                Ok(db_cid.into())
+            })
+            .collect()
+    }
+
+    pub fn write_db_ruv<I, J>(&mut self, mut added: I, mut removed: J) -> Result<(), OperationError>
+    where
+        I: Iterator<Item = Cid>,
+        J: Iterator<Item = Cid>,
+    {
+        let mut stmt = self
+            .get_conn()?
+            .prepare(&format!(
+                "DELETE FROM {}.ruv WHERE cid = :cid",
+                self.get_db_name()
+            ))
+            .map_err(sqlite_error)?;
+
+        removed.try_for_each(|cid| {
+            let db_cid: DbCidV1 = cid.into();
+
+            serde_json::to_string(&db_cid)
+                .map_err(serde_json_error)
+                .and_then(|ser_cid| {
+                    stmt.execute(named_params! {
+                        ":cid": &ser_cid
+                    })
+                    // remove the updated usize
+                    .map(|_| ())
+                    .map_err(sqlite_error)
+                })
+        })?;
+
+        let mut stmt = self
+            .get_conn()?
+            .prepare(&format!(
+                "INSERT OR REPLACE INTO {}.ruv (cid) VALUES(:cid)",
+                self.get_db_name()
+            ))
+            .map_err(sqlite_error)?;
+
+        added.try_for_each(|cid| {
+            let db_cid: DbCidV1 = cid.into();
+
+            serde_json::to_string(&db_cid)
+                .map_err(serde_json_error)
+                .and_then(|ser_cid| {
+                    stmt.execute(named_params! {
+                        ":cid": &ser_cid
+                    })
+                    // remove the updated usize
+                    .map(|_| ())
+                    .map_err(sqlite_error)
+                })
+        })
+    }
+
     pub fn create_idx(&self, attr: Attribute, itype: IndexType) -> Result<(), OperationError> {
         // Is there a better way than formatting this? I can't seem
         // to template into the str.
@@ -1629,7 +1725,19 @@ impl IdlSqliteWriteTransaction {
             dbv_id2entry = 8;
             info!(entry = %dbv_id2entry, "dbv_id2entry migrated (keyhandles)");
         }
-        //   * if v8 -> complete
+        //   * if v8 -> migrate all entries to have a change state
+        if dbv_id2entry == 8 {
+            self.migrate_dbentryv2_to_dbentryv3()?;
+            dbv_id2entry = 9;
+            info!(entry = %dbv_id2entry, "dbv_id2entry migrated (dbentryv2 -> dbentryv3)");
+        }
+        //   * if v9 -> complete
+        if dbv_id2entry == 9 {
+            self.create_db_ruv()?;
+            dbv_id2entry = 10;
+            info!(entry = %dbv_id2entry, "dbv_id2entry migrated (db_ruv)");
+        }
+        //   * if v10 -> complete
 
         self.set_db_version_key(DBV_ID2ENTRY, dbv_id2entry)?;
 
diff --git a/server/lib/src/be/keystorage.rs b/server/lib/src/be/keystorage.rs
index 54dee2206..a491f7903 100644
--- a/server/lib/src/be/keystorage.rs
+++ b/server/lib/src/be/keystorage.rs
@@ -91,7 +91,7 @@ impl<'a> IdlArcSqliteWriteTransaction<'a> {
     ) -> Result<(), OperationError> {
         self.db.set_key_handles(&keyhandles)?;
         self.keyhandles.clear();
-        self.keyhandles.extend(keyhandles.into_iter());
+        self.keyhandles.extend(keyhandles);
         Ok(())
     }
 }
diff --git a/server/lib/src/be/mod.rs b/server/lib/src/be/mod.rs
index 85ba4f869..b7436159e 100644
--- a/server/lib/src/be/mod.rs
+++ b/server/lib/src/be/mod.rs
@@ -20,6 +20,7 @@ use tracing::{trace, trace_span};
 use uuid::Uuid;
 
 use crate::be::dbentry::{DbBackup, DbEntry};
+use crate::be::dbrepl::DbReplMeta;
 use crate::entry::Entry;
 use crate::filter::{Filter, FilterPlan, FilterResolved, FilterValidResolved};
 use crate::prelude::*;
@@ -32,7 +33,9 @@ use crate::repl::ruv::{
 use crate::value::{IndexType, Value};
 
 pub mod dbentry;
+pub mod dbrepl;
 pub mod dbvalue;
+
 mod idl_arc_sqlite;
 mod idl_sqlite;
 pub(crate) mod idxkey;
@@ -847,6 +850,8 @@ pub trait BackendTransaction {
     }
 
     fn backup(&mut self, dst_path: &str) -> Result<(), OperationError> {
+        let repl_meta = self.get_ruv().to_db_backup_ruv();
+
         // load all entries into RAM, may need to change this later
         // if the size of the database compared to RAM is an issue
         let idl = IdList::AllIds;
@@ -875,11 +880,12 @@ pub trait BackendTransaction {
 
         let keyhandles = idlayer.get_key_handles()?;
 
-        let bak = DbBackup::V3 {
+        let bak = DbBackup::V4 {
             db_s_uuid,
             db_d_uuid,
             db_ts_max,
             keyhandles,
+            repl_meta,
             entries,
         };
 
@@ -1693,19 +1699,20 @@ impl<'a> BackendWriteTransaction<'a> {
     }
 
     pub fn restore(&mut self, src_path: &str) -> Result<(), OperationError> {
-        let idlayer = self.get_idlayer();
-        // load all entries into RAM, may need to change this later
-        // if the size of the database compared to RAM is an issue
         let serialized_string = fs::read_to_string(src_path).map_err(|e| {
             admin_error!("fs::read_to_string {:?}", e);
             OperationError::FsError
         })?;
 
-        idlayer.danger_purge_id2entry().map_err(|e| {
-            admin_error!("purge_id2entry failed {:?}", e);
+        self.danger_delete_all_db_content().map_err(|e| {
+            admin_error!("delete_all_db_content failed {:?}", e);
             e
         })?;
 
+        let idlayer = self.get_idlayer();
+        // load all entries into RAM, may need to change this later
+        // if the size of the database compared to RAM is an issue
+
         let dbbak_option: Result<DbBackup, serde_json::Error> =
             serde_json::from_str(&serialized_string);
 
@@ -1714,8 +1721,8 @@ impl<'a> BackendWriteTransaction<'a> {
             OperationError::SerdeJsonError
         })?;
 
-        let dbentries = match dbbak {
-            DbBackup::V1(dbentries) => dbentries,
+        let (dbentries, repl_meta) = match dbbak {
+            DbBackup::V1(dbentries) => (dbentries, None),
             DbBackup::V2 {
                 db_s_uuid,
                 db_d_uuid,
@@ -1726,7 +1733,7 @@ impl<'a> BackendWriteTransaction<'a> {
                 idlayer.write_db_s_uuid(db_s_uuid)?;
                 idlayer.write_db_d_uuid(db_d_uuid)?;
                 idlayer.set_db_ts_max(db_ts_max)?;
-                entries
+                (entries, None)
             }
             DbBackup::V3 {
                 db_s_uuid,
@@ -1740,10 +1747,36 @@ impl<'a> BackendWriteTransaction<'a> {
                 idlayer.write_db_d_uuid(db_d_uuid)?;
                 idlayer.set_db_ts_max(db_ts_max)?;
                 idlayer.set_key_handles(keyhandles)?;
-                entries
+                (entries, None)
+            }
+            DbBackup::V4 {
+                db_s_uuid,
+                db_d_uuid,
+                db_ts_max,
+                keyhandles,
+                repl_meta,
+                entries,
+            } => {
+                // Do stuff.
+                idlayer.write_db_s_uuid(db_s_uuid)?;
+                idlayer.write_db_d_uuid(db_d_uuid)?;
+                idlayer.set_db_ts_max(db_ts_max)?;
+                idlayer.set_key_handles(keyhandles)?;
+                (entries, Some(repl_meta))
             }
         };
 
+        // Rebuild the RUV from the backup.
+        match repl_meta {
+            Some(DbReplMeta::V1 { ruv: db_ruv }) => {
+                self.get_ruv()
+                    .restore(db_ruv.into_iter().map(|db_cid| db_cid.into()))?;
+            }
+            None => {
+                warn!("Unable to restore replication metadata, this server may need a refresh.");
+            }
+        }
+
         info!("Restoring {} entries ...", dbentries.len());
 
         // Migrate any v1 entries to v2 if needed.
@@ -1763,6 +1796,8 @@ impl<'a> BackendWriteTransaction<'a> {
             })
             .collect();
 
+        let idlayer = self.get_idlayer();
+
         idlayer.write_identries_raw(identries?.into_iter())?;
 
         info!("Restored {} entries", dbentries.len());
@@ -1778,8 +1813,33 @@ impl<'a> BackendWriteTransaction<'a> {
         }
     }
 
+    /// If any RUV elements are present in the DB, load them now. This provides us with
+    /// the RUV boundaries and change points from previous operations of the server, so
+    /// that ruv_rebuild can "fill in" the gaps.
+    ///
+    /// # SAFETY
+    ///
+    /// Note that you should only call this function during the server startup
+    /// to reload the RUV data from the entries of the database.
+    ///
+    /// Before calling this, the in memory ruv MUST be clear.
     #[instrument(level = "debug", name = "be::ruv_rebuild", skip_all)]
-    pub fn ruv_rebuild(&mut self) -> Result<(), OperationError> {
+    fn ruv_reload(&mut self) -> Result<(), OperationError> {
+        let idlayer = self.get_idlayer();
+
+        let db_ruv = idlayer.get_db_ruv()?;
+
+        // Setup the CID's that existed previously. We don't need to know what entries
+        // they affect, we just need them to ensure that we have ranges for replication
+        // comparison to take effect properly.
+        self.get_ruv().restore(db_ruv)?;
+
+        // Then populate the RUV with the data from the entries.
+        self.ruv_rebuild()
+    }
+
+    #[instrument(level = "debug", name = "be::ruv_rebuild", skip_all)]
+    fn ruv_rebuild(&mut self) -> Result<(), OperationError> {
         // Rebuild the ruv!
         // For now this has to read from all the entries in the DB, but in the future
         // we'll actually store this properly (?). If it turns out this is really fast
@@ -1819,11 +1879,14 @@ impl<'a> BackendWriteTransaction<'a> {
 
     pub fn commit(self) -> Result<(), OperationError> {
         let BackendWriteTransaction {
-            idlayer,
+            mut idlayer,
             idxmeta_wr,
             ruv,
         } = self;
 
+        // write the ruv content back to the db.
+        idlayer.write_db_ruv(ruv.added(), ruv.removed())?;
+
         idlayer.commit().map(|()| {
             ruv.commit();
             idxmeta_wr.commit();
@@ -1975,7 +2038,7 @@ impl Backend {
         // Now rebuild the ruv.
         let mut be_write = be.write();
         be_write
-            .ruv_rebuild()
+            .ruv_reload()
             .and_then(|_| be_write.commit())
             .map_err(|e| {
                 admin_error!(?e, "Failed to reload ruv");
@@ -2532,6 +2595,16 @@ mod tests {
                 } => {
                     let _ = entries.pop();
                 }
+                DbBackup::V4 {
+                    db_s_uuid: _,
+                    db_d_uuid: _,
+                    db_ts_max: _,
+                    keyhandles: _,
+                    repl_meta: _,
+                    entries,
+                } => {
+                    let _ = entries.pop();
+                }
             };
 
             let serialized_entries_str = serde_json::to_string_pretty(&dbbak).unwrap();
diff --git a/server/lib/src/constants/entries.rs b/server/lib/src/constants/entries.rs
index 3174b7880..fc6e73c87 100644
--- a/server/lib/src/constants/entries.rs
+++ b/server/lib/src/constants/entries.rs
@@ -805,7 +805,7 @@ lazy_static! {
         name: "idm_people_extend_priv",
         description: "Builtin System Administrators Group.",
         uuid: UUID_IDM_PEOPLE_EXTEND_PRIV,
-        members: vec![BUILTIN_ACCOUNT_ADMIN.uuid],
+        members: vec![UUID_SYSTEM_ADMINS],
         ..Default::default()
     };
     /// Self-write of mail
@@ -840,7 +840,7 @@ lazy_static! {
         name: "idm_hp_people_extend_priv",
         description: "Builtin IDM Group for extending high privilege accounts to be people.",
         uuid: UUID_IDM_HP_PEOPLE_EXTEND_PRIV,
-        members: vec![BUILTIN_ACCOUNT_ADMIN.uuid],
+        members: vec![UUID_SYSTEM_ADMINS],
         ..Default::default()
     };
 
diff --git a/server/lib/src/entry.rs b/server/lib/src/entry.rs
index 9d3e63217..8b07805ee 100644
--- a/server/lib/src/entry.rs
+++ b/server/lib/src/entry.rs
@@ -45,7 +45,7 @@ use tracing::trace;
 use uuid::Uuid;
 use webauthn_rs::prelude::{AttestedPasskey as DeviceKeyV4, Passkey as PasskeyV4};
 
-use crate::be::dbentry::{DbEntry, DbEntryV2, DbEntryVers};
+use crate::be::dbentry::{DbEntry, DbEntryVers};
 use crate::be::dbvalue::DbValueSetV2;
 use crate::be::{IdxKey, IdxSlope};
 use crate::credential::Credential;
@@ -904,28 +904,28 @@ impl Entry<EntryIncremental, EntryNew> {
 
                             match (self.attrs.get(attr_name), db_ent.attrs.get(attr_name)) {
                                 (Some(vs_left), Some(vs_right)) if take_left => {
+                                    changes.insert(attr_name.clone(), cid_left.clone());
                                     #[allow(clippy::todo)]
-                                    if let Some(_attr_state) =
+                                    if let Some(merged_attr_state) =
                                         vs_left.repl_merge_valueset(vs_right, trim_cid)
                                     {
-                                        // TODO note: This is for special attr types that need to merge
+                                        // NOTE: This is for special attr types that need to merge
                                         // rather than choose content.
-                                        todo!();
+                                        eattrs.insert(attr_name.clone(), merged_attr_state);
                                     } else {
-                                        changes.insert(attr_name.clone(), cid_left.clone());
                                         eattrs.insert(attr_name.clone(), vs_left.clone());
                                     }
                                 }
                                 (Some(vs_left), Some(vs_right)) => {
+                                    changes.insert(attr_name.clone(), cid_right.clone());
                                     #[allow(clippy::todo)]
-                                    if let Some(_attr_state) =
+                                    if let Some(merged_attr_state) =
                                         vs_right.repl_merge_valueset(vs_left, trim_cid)
                                     {
-                                        // TODO note: This is for special attr types that need to merge
+                                        // NOTE: This is for special attr types that need to merge
                                         // rather than choose content.
-                                        todo!();
+                                        eattrs.insert(attr_name.clone(), merged_attr_state);
                                     } else {
-                                        changes.insert(attr_name.clone(), cid_right.clone());
                                         eattrs.insert(attr_name.clone(), vs_right.clone());
                                     }
                                 }
@@ -1368,14 +1368,9 @@ impl Entry<EntrySealed, EntryCommitted> {
     pub fn to_dbentry(&self) -> DbEntry {
         // In the future this will do extra work to process uuid
         // into "attributes" suitable for dbentry storage.
-
-        // How will this work with replication?
-        //
-        // Alternately, we may have higher-level types that translate entry
-        // into proper structures, and they themself emit/modify entries?
-
         DbEntry {
-            ent: DbEntryVers::V2(DbEntryV2 {
+            ent: DbEntryVers::V3 {
+                changestate: self.valid.ecstate.to_db_changestate(),
                 attrs: self
                     .attrs
                     .iter()
@@ -1384,7 +1379,7 @@ impl Entry<EntrySealed, EntryCommitted> {
                         (k.clone(), dbvs)
                     })
                     .collect(),
-            }),
+            },
         }
     }
 
@@ -1825,50 +1820,75 @@ impl Entry<EntrySealed, EntryCommitted> {
 
     pub fn from_dbentry(db_e: DbEntry, id: u64) -> Option<Self> {
         // Convert attrs from db format to value
-        let r_attrs: Result<Eattrs, ()> = match db_e.ent {
-            DbEntryVers::V1(_) => {
-                admin_error!("Db V1 entry should have been migrated!");
-                Err(())
-            }
-            DbEntryVers::V2(v2) => v2
-                .attrs
-                .into_iter()
-                // Skip anything empty as new VS can't deal with it.
-                .filter(|(_k, vs)| !vs.is_empty())
-                .map(|(k, dbvs)| {
-                    valueset::from_db_valueset_v2(dbvs)
-                        .map(|vs: ValueSet| (k, vs))
-                        .map_err(|e| {
-                            admin_error!(?e, "from_dbentry failed");
-                        })
-                })
-                .collect(),
-        };
 
-        let attrs = r_attrs.ok()?;
+        let (attrs, ecstate) = match db_e.ent {
+            DbEntryVers::V1(_) => {
+                error!("Db V1 entry should have been migrated!");
+                return None;
+            }
+            DbEntryVers::V2(v2) => {
+                let r_attrs = v2
+                    .attrs
+                    .into_iter()
+                    // Skip anything empty as new VS can't deal with it.
+                    .filter(|(_k, vs)| !vs.is_empty())
+                    .map(|(k, dbvs)| {
+                        valueset::from_db_valueset_v2(dbvs)
+                            .map(|vs: ValueSet| (k, vs))
+                            .map_err(|e| {
+                                error!(?e, "from_dbentry failed");
+                            })
+                    })
+                    .collect::<Result<Eattrs, ()>>()
+                    .ok()?;
+
+                /*
+                 * ⚠️  ==== The Hack Zoen ==== ⚠️
+                 *
+                 * For now to make replication work, we are synthesising an in-memory change
+                 * log, pinned to "the last time the entry was modified" as it's "create time".
+                 *
+                 * This should only be done *once* on entry load.
+                 */
+                let cid = r_attrs
+                    .get(Attribute::LastModifiedCid.as_ref())
+                    .and_then(|vs| vs.as_cid_set())
+                    .and_then(|set| set.iter().next().cloned())
+                    .or_else(|| {
+                        error!("Unable to access last modified cid of entry, unable to proceed");
+                        None
+                    })?;
+
+                let ecstate = EntryChangeState::new_without_schema(&cid, &r_attrs);
+
+                (r_attrs, ecstate)
+            }
+
+            DbEntryVers::V3 { changestate, attrs } => {
+                let ecstate = EntryChangeState::from_db_changestate(changestate);
+
+                let r_attrs = attrs
+                    .into_iter()
+                    // Skip anything empty as new VS can't deal with it.
+                    .filter(|(_k, vs)| !vs.is_empty())
+                    .map(|(k, dbvs)| {
+                        valueset::from_db_valueset_v2(dbvs)
+                            .map(|vs: ValueSet| (k, vs))
+                            .map_err(|e| {
+                                error!(?e, "from_dbentry failed");
+                            })
+                    })
+                    .collect::<Result<Eattrs, ()>>()
+                    .ok()?;
+
+                (r_attrs, ecstate)
+            }
+        };
 
         let uuid = attrs
             .get(Attribute::Uuid.as_ref())
             .and_then(|vs| vs.to_uuid_single())?;
 
-        /*
-         * ⚠️  ==== The Hack Zoen ==== ⚠️
-         *
-         * For now to make replication work, we are synthesising an in-memory change
-         * log, pinned to "the last time the entry was modified" as it's "create time".
-         *
-         * This means that a simple restart of the server flushes and resets the cl
-         * content. In the future though, this will actually be "part of" the entry
-         * and loaded from disk proper.
-         */
-        let cid = attrs
-            .get(Attribute::LastModifiedCid.as_ref())
-            .and_then(|vs| vs.as_cid_set())
-            .and_then(|set| set.iter().next().cloned())?;
-
-        // let eclog = EntryChangelog::new_without_schema(cid, attrs.clone());
-        let ecstate = EntryChangeState::new_without_schema(&cid, &attrs);
-
         Some(Entry {
             valid: EntrySealed { uuid, ecstate },
             state: EntryCommitted { id },
diff --git a/server/lib/src/repl/cid.rs b/server/lib/src/repl/cid.rs
index 197c97041..33f90c2c6 100644
--- a/server/lib/src/repl/cid.rs
+++ b/server/lib/src/repl/cid.rs
@@ -1,6 +1,7 @@
 use std::fmt;
 use std::time::Duration;
 
+use crate::be::dbvalue::DbCidV1;
 use crate::prelude::*;
 use kanidm_proto::v1::OperationError;
 use serde::{Deserialize, Serialize};
@@ -12,6 +13,20 @@ pub struct Cid {
     pub s_uuid: Uuid,
 }
 
+impl From<DbCidV1> for Cid {
+    fn from(
+        DbCidV1 {
+            server_id,
+            timestamp,
+        }: DbCidV1,
+    ) -> Self {
+        Cid {
+            ts: timestamp,
+            s_uuid: server_id,
+        }
+    }
+}
+
 impl fmt::Display for Cid {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         write!(f, "{:032}-{}", self.ts.as_nanos(), self.s_uuid)
diff --git a/server/lib/src/repl/entry.rs b/server/lib/src/repl/entry.rs
index f2795a71f..b5d8adb02 100644
--- a/server/lib/src/repl/entry.rs
+++ b/server/lib/src/repl/entry.rs
@@ -1,4 +1,6 @@
 use super::cid::Cid;
+use crate::be::dbrepl::DbEntryChangeState;
+use crate::be::dbvalue::DbCidV1;
 use crate::entry::Eattrs;
 use crate::prelude::*;
 use crate::schema::SchemaTransaction;
@@ -61,6 +63,76 @@ impl EntryChangeState {
         EntryChangeState { st }
     }
 
+    pub(crate) fn to_db_changestate(&self) -> DbEntryChangeState {
+        match &self.st {
+            State::Live { at, changes } => {
+                let at = DbCidV1 {
+                    server_id: at.s_uuid,
+                    timestamp: at.ts,
+                };
+
+                let changes = changes
+                    .iter()
+                    .map(|(attr, cid)| {
+                        (
+                            attr.to_string(),
+                            DbCidV1 {
+                                server_id: cid.s_uuid,
+                                timestamp: cid.ts,
+                            },
+                        )
+                    })
+                    .collect();
+
+                DbEntryChangeState::V1Live { at, changes }
+            }
+            State::Tombstone { at } => {
+                let at = DbCidV1 {
+                    server_id: at.s_uuid,
+                    timestamp: at.ts,
+                };
+
+                DbEntryChangeState::V1Tombstone { at }
+            }
+        }
+    }
+
+    pub(crate) fn from_db_changestate(db_ecstate: DbEntryChangeState) -> Self {
+        match db_ecstate {
+            DbEntryChangeState::V1Live { at, changes } => {
+                let at = Cid {
+                    s_uuid: at.server_id,
+                    ts: at.timestamp,
+                };
+
+                let changes = changes
+                    .iter()
+                    .map(|(attr, cid)| {
+                        (
+                            attr.into(),
+                            Cid {
+                                s_uuid: cid.server_id,
+                                ts: cid.timestamp,
+                            },
+                        )
+                    })
+                    .collect();
+
+                EntryChangeState {
+                    st: State::Live { at, changes },
+                }
+            }
+            DbEntryChangeState::V1Tombstone { at } => EntryChangeState {
+                st: State::Tombstone {
+                    at: Cid {
+                        s_uuid: at.server_id,
+                        ts: at.timestamp,
+                    },
+                },
+            },
+        }
+    }
+
     pub(crate) fn build(st: State) -> Self {
         EntryChangeState { st }
     }
diff --git a/server/lib/src/repl/ruv.rs b/server/lib/src/repl/ruv.rs
index cac2f202a..720c2df0f 100644
--- a/server/lib/src/repl/ruv.rs
+++ b/server/lib/src/repl/ruv.rs
@@ -1,3 +1,4 @@
+use crate::be::dbrepl::DbReplMeta;
 use std::cmp::Ordering;
 use std::collections::{BTreeMap, BTreeSet};
 use std::ops::Bound::*;
@@ -5,6 +6,7 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use concread::bptree::{BptreeMap, BptreeMapReadSnapshot, BptreeMapReadTxn, BptreeMapWriteTxn};
+
 use idlset::v2::IDLBitRange;
 use kanidm_proto::v1::ConsistencyError;
 
@@ -60,7 +62,10 @@ pub(crate) enum RangeDiffStatus {
 impl ReplicationUpdateVector {
     pub fn write(&self) -> ReplicationUpdateVectorWriteTransaction<'_> {
         ReplicationUpdateVectorWriteTransaction {
+            // Need to take the write first.
+            cleared: false,
             data: self.data.write(),
+            data_pre: self.data.read(),
             ranged: self.ranged.write(),
         }
     }
@@ -212,16 +217,22 @@ impl ReplicationUpdateVector {
 }
 
 pub struct ReplicationUpdateVectorWriteTransaction<'a> {
+    cleared: bool,
     data: BptreeMapWriteTxn<'a, Cid, IDLBitRange>,
+    data_pre: BptreeMapReadTxn<'a, Cid, IDLBitRange>,
     ranged: BptreeMapWriteTxn<'a, Uuid, BTreeSet<Duration>>,
 }
 
 impl<'a> fmt::Debug for ReplicationUpdateVectorWriteTransaction<'a> {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        writeln!(f, "RUV DUMP")?;
+        writeln!(f, "RUV DATA DUMP")?;
         self.data
             .iter()
-            .try_for_each(|(cid, idl)| writeln!(f, "* [{cid} {idl:?}]"))
+            .try_for_each(|(cid, idl)| writeln!(f, "* [{cid} {idl:?}]"))?;
+        writeln!(f, "RUV RANGE DUMP")?;
+        self.ranged
+            .iter()
+            .try_for_each(|(s_uuid, ts)| writeln!(f, "* [{s_uuid} {ts:?}]"))
     }
 }
 
@@ -230,11 +241,30 @@ pub struct ReplicationUpdateVectorReadTransaction<'a> {
     ranged: BptreeMapReadTxn<'a, Uuid, BTreeSet<Duration>>,
 }
 
+impl<'a> fmt::Debug for ReplicationUpdateVectorReadTransaction<'a> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        writeln!(f, "RUV DATA DUMP")?;
+        self.data
+            .iter()
+            .try_for_each(|(cid, idl)| writeln!(f, "* [{cid} {idl:?}]"))?;
+        writeln!(f, "RUV RANGE DUMP")?;
+        self.ranged
+            .iter()
+            .try_for_each(|(s_uuid, ts)| writeln!(f, "* [{s_uuid} {ts:?}]"))
+    }
+}
+
 pub trait ReplicationUpdateVectorTransaction {
     fn ruv_snapshot(&self) -> BptreeMapReadSnapshot<'_, Cid, IDLBitRange>;
 
     fn range_snapshot(&self) -> BptreeMapReadSnapshot<'_, Uuid, BTreeSet<Duration>>;
 
+    fn to_db_backup_ruv(&self) -> DbReplMeta {
+        DbReplMeta::V1 {
+            ruv: self.ruv_snapshot().keys().map(|cid| cid.into()).collect(),
+        }
+    }
+
     fn current_ruv_range(&self) -> Result<BTreeMap<Uuid, ReplCidRange>, OperationError> {
         self.range_snapshot()
             .iter()
@@ -447,6 +477,7 @@ impl<'a> ReplicationUpdateVectorTransaction for ReplicationUpdateVectorReadTrans
 
 impl<'a> ReplicationUpdateVectorWriteTransaction<'a> {
     pub fn clear(&mut self) {
+        self.cleared = true;
         self.data.clear();
         self.ranged.clear();
     }
@@ -524,10 +555,42 @@ impl<'a> ReplicationUpdateVectorWriteTransaction<'a> {
         Ok(())
     }
 
+    /// Restore the ruv from a DB backup. It's important to note here that
+    /// we don't actually need to restore and of the IDL's in the process. we only
+    /// needs the CID's of the changes/points in time. This is because when the
+    /// db entries are restored, their changesets will re-populate the data that we
+    /// need in the RUV at these points. The reason we need these ranges without IDL
+    /// is so that trim and replication works properly.
+    pub(crate) fn restore<I>(&mut self, iter: I) -> Result<(), OperationError>
+    where
+        I: IntoIterator<Item = Cid>,
+    {
+        let mut rebuild_ruv: BTreeMap<Cid, IDLBitRange> = BTreeMap::new();
+        let mut rebuild_range: BTreeMap<Uuid, BTreeSet<Duration>> = BTreeMap::default();
+
+        for cid in iter {
+            if !rebuild_ruv.contains_key(&cid) {
+                let idl = IDLBitRange::new();
+                rebuild_ruv.insert(cid.clone(), idl);
+            }
+
+            if let Some(server_range) = rebuild_range.get_mut(&cid.s_uuid) {
+                server_range.insert(cid.ts);
+            } else {
+                let mut ts_range = BTreeSet::default();
+                ts_range.insert(cid.ts);
+                rebuild_range.insert(cid.s_uuid, ts_range);
+            }
+        }
+
+        self.data.extend(rebuild_ruv);
+        self.ranged.extend(rebuild_range);
+
+        Ok(())
+    }
+
     pub fn rebuild(&mut self, entries: &[Arc<EntrySealedCommitted>]) -> Result<(), OperationError> {
-        // Drop everything.
-        self.clear();
-        // Entries and their internal changelogs are the "source of truth" for all changes
+        // Entries and their internal changestates are the "source of truth" for all changes
         // that have ever occurred and are stored on this server. So we use them to rebuild our RUV
         // here!
         let mut rebuild_ruv: BTreeMap<Cid, IDLBitRange> = BTreeMap::new();
@@ -749,6 +812,8 @@ impl<'a> ReplicationUpdateVectorWriteTransaction<'a> {
         }
 
         /*
+        // For now, we can't actually remove any server_id's because we would break range
+        // comparisons in this case. We likely need a way to clean-ruv here.
         for s_uuid in remove_suuid {
             let x = self.ranged.remove(&s_uuid);
             assert!(x.map(|y| y.is_empty()).unwrap_or(false))
@@ -762,6 +827,49 @@ impl<'a> ReplicationUpdateVectorWriteTransaction<'a> {
         Ok(idl)
     }
 
+    pub fn added(&self) -> impl Iterator<Item = Cid> + '_ {
+        // Find the max from the previous dataset.
+        let prev_bound = if self.cleared {
+            // We have been cleared during this txn, so everything in data is
+            // added.
+            Unbounded
+        } else if let Some((max, _)) = self.data_pre.last_key_value() {
+            Excluded(max.clone())
+        } else {
+            // If empty, assume everything is new.
+            Unbounded
+        };
+
+        // Starting from the previous max, iterate through our data to find what
+        // has been added.
+        self.data.range((prev_bound, Unbounded)).map(|(cid, _)| {
+            trace!(added_cid = ?cid);
+            cid.clone()
+        })
+    }
+
+    pub fn removed(&self) -> impl Iterator<Item = Cid> + '_ {
+        let prev_bound = if self.cleared {
+            // We have been cleared during this txn, so everything in pre is
+            // removed.
+            Unbounded
+        } else if let Some((min, _)) = self.data.first_key_value() {
+            Excluded(min.clone())
+        } else {
+            // If empty, assume everything is new.
+            Unbounded
+        };
+
+        // iterate through our previous data to find what has been removed given
+        // the ranges determined above.
+        self.data_pre
+            .range((Unbounded, prev_bound))
+            .map(|(cid, _)| {
+                trace!(removed_cid = ?cid);
+                cid.clone()
+            })
+    }
+
     pub fn commit(self) {
         self.data.commit();
         self.ranged.commit();
diff --git a/server/lib/src/repl/supplier.rs b/server/lib/src/repl/supplier.rs
index a9d939be8..539027bf3 100644
--- a/server/lib/src/repl/supplier.rs
+++ b/server/lib/src/repl/supplier.rs
@@ -10,10 +10,12 @@ use kanidm_lib_crypto::mtls::build_self_signed_server_and_client_identity;
 use kanidm_lib_crypto::prelude::{PKey, Private, X509};
 
 impl<'a> QueryServerWriteTransaction<'a> {
-    fn supplier_generate_key_cert(&mut self) -> Result<(PKey<Private>, X509), OperationError> {
+    fn supplier_generate_key_cert(
+        &mut self,
+        domain_name: &str,
+    ) -> Result<(PKey<Private>, X509), OperationError> {
         // Invalid, must need to re-generate.
-        let domain_name = "localhost";
-        let expiration_days = 1;
+        let expiration_days = 180;
         let s_uuid = self.get_server_uuid();
 
         let (private, x509) =
@@ -39,12 +41,15 @@ impl<'a> QueryServerWriteTransaction<'a> {
     }
 
     #[instrument(level = "info", skip_all)]
-    pub fn supplier_renew_key_cert(&mut self) -> Result<(), OperationError> {
-        self.supplier_generate_key_cert().map(|_| ())
+    pub fn supplier_renew_key_cert(&mut self, domain_name: &str) -> Result<(), OperationError> {
+        self.supplier_generate_key_cert(domain_name).map(|_| ())
     }
 
     #[instrument(level = "info", skip_all)]
-    pub fn supplier_get_key_cert(&mut self) -> Result<(PKey<Private>, X509), OperationError> {
+    pub fn supplier_get_key_cert(
+        &mut self,
+        domain_name: &str,
+    ) -> Result<(PKey<Private>, X509), OperationError> {
         // Later we need to put this through a HSM or similar, but we will always need a way
         // to persist a handle, so we still need the db write and load components.
 
@@ -66,7 +71,7 @@ impl<'a> QueryServerWriteTransaction<'a> {
                 // error? regenerate?
             }
             */
-            None => self.supplier_generate_key_cert()?,
+            None => self.supplier_generate_key_cert(domain_name)?,
         };
 
         Ok(key_cert)
@@ -100,14 +105,12 @@ impl<'a> QueryServerReadTransaction<'a> {
             return Ok(ReplIncrementalContext::DomainMismatch);
         }
 
-        let our_ranges = self
-            .get_be_txn()
-            .get_ruv()
-            .current_ruv_range()
-            .map_err(|e| {
-                error!(err = ?e, "Unable to access supplier RUV range");
-                e
-            })?;
+        let supplier_ruv = self.get_be_txn().get_ruv();
+
+        let our_ranges = supplier_ruv.current_ruv_range().map_err(|e| {
+            error!(err = ?e, "Unable to access supplier RUV range");
+            e
+        })?;
 
         // Compare this to our internal ranges - work out the list of entry
         // id's that are now different.
@@ -120,21 +123,27 @@ impl<'a> QueryServerReadTransaction<'a> {
             RangeDiffStatus::Ok(ranges) => ranges,
             RangeDiffStatus::Refresh { lag_range } => {
                 error!("Replication - Consumer is lagging and must be refreshed.");
-                debug!(?lag_range);
+                info!(?lag_range);
+                debug!(consumer_ranges = ?ctx_ranges);
+                debug!(supplier_ranges = ?our_ranges);
                 return Ok(ReplIncrementalContext::RefreshRequired);
             }
             RangeDiffStatus::Unwilling { adv_range } => {
                 error!("Replication - Supplier is lagging and must be investigated.");
-                debug!(?adv_range);
+                info!(?adv_range);
+                debug!(consumer_ranges = ?ctx_ranges);
+                debug!(supplier_ranges = ?our_ranges);
                 return Ok(ReplIncrementalContext::UnwillingToSupply);
             }
             RangeDiffStatus::Critical {
                 lag_range,
                 adv_range,
             } => {
-                error!("Replication Critical - Servers are advanced of us, and also lagging! This must be immediately investigated!");
-                debug!(?lag_range);
-                debug!(?adv_range);
+                error!("Replication Critical - Consumers are advanced of us, and also lagging! This must be immediately investigated!");
+                info!(?lag_range);
+                info!(?adv_range);
+                debug!(consumer_ranges = ?ctx_ranges);
+                debug!(supplier_ranges = ?our_ranges);
                 return Ok(ReplIncrementalContext::UnwillingToSupply);
             }
         };
diff --git a/server/lib/src/repl/tests.rs b/server/lib/src/repl/tests.rs
index 5a54ebb59..044ef408f 100644
--- a/server/lib/src/repl/tests.rs
+++ b/server/lib/src/repl/tests.rs
@@ -1,11 +1,15 @@
 use crate::be::BackendTransaction;
+use crate::credential::Credential;
 use crate::prelude::*;
 use crate::repl::entry::State;
 use crate::repl::proto::ConsumerState;
 use crate::repl::proto::ReplIncrementalContext;
 use crate::repl::ruv::ReplicationUpdateVectorTransaction;
 use crate::repl::ruv::{RangeDiffStatus, ReplicationUpdateVector};
+use crate::value::{Session, SessionState};
+use kanidm_lib_crypto::CryptoPolicy;
 use std::collections::BTreeMap;
+use time::OffsetDateTime;
 
 fn repl_initialise(
     from: &mut QueryServerReadTransaction<'_>,
@@ -2981,6 +2985,188 @@ async fn test_repl_initial_consumer_join(server_a: &QueryServer, server_b: &Quer
     drop(server_b_txn);
 }
 
+// Test handling of sessions over replication
+
+#[qs_pair_test]
+async fn test_repl_increment_session_new(server_a: &QueryServer, server_b: &QueryServer) {
+    let ct = duration_from_epoch_now();
+
+    // First create a user.
+
+    let mut server_b_txn = server_b.write(ct).await;
+
+    let t_uuid = Uuid::new_v4();
+
+    let p = CryptoPolicy::minimum();
+    let cred = Credential::new_password_only(&p, "test_password").unwrap();
+    let cred_id = cred.uuid;
+
+    let e1 = entry_init!(
+        (Attribute::Class, EntryClass::Object.to_value()),
+        (Attribute::Class, EntryClass::Person.to_value()),
+        (Attribute::Class, EntryClass::Account.to_value()),
+        (Attribute::Name, Value::new_iname("testperson1")),
+        (Attribute::Uuid, Value::Uuid(t_uuid)),
+        (Attribute::Description, Value::new_utf8s("testperson1")),
+        (Attribute::DisplayName, Value::new_utf8s("testperson1")),
+        (
+            Attribute::PrimaryCredential,
+            Value::Cred("primary".to_string(), cred.clone())
+        )
+    );
+
+    let ce = CreateEvent::new_internal(vec![e1]);
+    assert!(server_b_txn.create(&ce).is_ok());
+
+    server_b_txn.commit().expect("Failed to commit");
+
+    let mut server_a_txn = server_a.write(ct).await;
+    let mut server_b_txn = server_b.read().await;
+
+    assert!(repl_initialise(&mut server_b_txn, &mut server_a_txn)
+        .and_then(|_| server_a_txn.commit())
+        .is_ok());
+    drop(server_b_txn);
+
+    // Update a session on A.
+
+    let mut server_a_txn = server_a.write(ct).await;
+
+    let curtime_odt = OffsetDateTime::UNIX_EPOCH + ct;
+    let exp_curtime = ct + Duration::from_secs(60);
+    let exp_curtime_odt = OffsetDateTime::UNIX_EPOCH + exp_curtime;
+
+    // Create a fake session.
+    let session_id_a = Uuid::new_v4();
+    let state = SessionState::ExpiresAt(exp_curtime_odt);
+    let issued_at = curtime_odt;
+    let issued_by = IdentityId::User(t_uuid);
+    let scope = SessionScope::ReadOnly;
+
+    let session = Value::Session(
+        session_id_a,
+        Session {
+            label: "label".to_string(),
+            state,
+            issued_at,
+            issued_by,
+            cred_id,
+            scope,
+        },
+    );
+
+    let modlist = ModifyList::new_append(Attribute::UserAuthTokenSession.into(), session);
+
+    server_a_txn
+        .internal_modify(
+            &filter!(f_eq(Attribute::Uuid, PartialValue::Uuid(t_uuid))),
+            &modlist,
+        )
+        .expect("Failed to modify user");
+
+    server_a_txn.commit().expect("Failed to commit");
+
+    // And a session on B.
+
+    let ct = duration_from_epoch_now();
+    let mut server_b_txn = server_b.write(ct).await;
+
+    let curtime_odt = OffsetDateTime::UNIX_EPOCH + ct;
+    let exp_curtime = ct + Duration::from_secs(60);
+    let exp_curtime_odt = OffsetDateTime::UNIX_EPOCH + exp_curtime;
+
+    // Create a fake session.
+    let session_id_b = Uuid::new_v4();
+    let state = SessionState::ExpiresAt(exp_curtime_odt);
+    let issued_at = curtime_odt;
+    let issued_by = IdentityId::User(t_uuid);
+    let scope = SessionScope::ReadOnly;
+
+    let session = Value::Session(
+        session_id_b,
+        Session {
+            label: "label".to_string(),
+            state,
+            issued_at,
+            issued_by,
+            cred_id,
+            scope,
+        },
+    );
+
+    let modlist = ModifyList::new_append(Attribute::UserAuthTokenSession.into(), session);
+
+    server_b_txn
+        .internal_modify(
+            &filter!(f_eq(Attribute::Uuid, PartialValue::Uuid(t_uuid))),
+            &modlist,
+        )
+        .expect("Failed to modify user");
+
+    server_b_txn.commit().expect("Failed to commit");
+
+    // Now incremental in both directions.
+
+    let ct = duration_from_epoch_now();
+    let mut server_a_txn = server_a.read().await;
+    let mut server_b_txn = server_b.write(ct).await;
+
+    trace!("========================================");
+    repl_incremental(&mut server_a_txn, &mut server_b_txn);
+
+    let e1 = server_a_txn
+        .internal_search_all_uuid(t_uuid)
+        .expect("Unable to access entry.");
+    let e2 = server_b_txn
+        .internal_search_all_uuid(t_uuid)
+        .expect("Unable to access entry.");
+
+    // Here, A's session should have merged with B.
+    let sessions_a = e1
+        .get_ava_as_session_map(Attribute::UserAuthTokenSession)
+        .unwrap();
+
+    assert!(sessions_a.len() == 1);
+    assert!(sessions_a.get(&session_id_a).is_some());
+    assert!(sessions_a.get(&session_id_b).is_none());
+
+    let sessions_b = e2
+        .get_ava_as_session_map(Attribute::UserAuthTokenSession)
+        .unwrap();
+
+    assert!(sessions_b.len() == 2);
+    assert!(sessions_b.get(&session_id_a).is_some());
+    assert!(sessions_b.get(&session_id_b).is_some());
+
+    server_b_txn.commit().expect("Failed to commit");
+    drop(server_a_txn);
+
+    let ct = duration_from_epoch_now();
+    let mut server_b_txn = server_b.read().await;
+    let mut server_a_txn = server_a.write(ct).await;
+
+    trace!("========================================");
+    repl_incremental(&mut server_b_txn, &mut server_a_txn);
+
+    let e1 = server_a_txn
+        .internal_search_all_uuid(t_uuid)
+        .expect("Unable to access entry.");
+    let e2 = server_b_txn
+        .internal_search_all_uuid(t_uuid)
+        .expect("Unable to access entry.");
+
+    let e1_cs = e1.get_changestate();
+    let e2_cs = e2.get_changestate();
+
+    assert!(e1_cs == e2_cs);
+    trace!(?e1);
+    trace!(?e2);
+    assert!(e1 == e2);
+
+    server_a_txn.commit().expect("Failed to commit");
+    drop(server_b_txn);
+}
+
 // Test change of domain version over incremental.
 //
 // todo when I have domain version migrations working.