Compare commits

...

135 Commits

Author SHA1 Message Date
Ginger f746a54e21 chore: Update ruwuma 2025-11-14 10:08:30 -05:00
Ginger a6e5dfcd50 chore(sync/v3): Remove unused imports 2025-11-14 09:20:51 -05:00
Ginger 51aa173a6e fix(sync/v3): Don't send rejected invites on initial syncs 2025-11-13 13:27:52 -05:00
Ginger 56fcc5d569 refactor(sync/v3): Extract left room timeline logic into its own function 2025-11-13 12:44:13 -05:00
Ginger 901bb63ecc fix(sync/v3): Don't send dummy leaves on an initial sync 2025-11-13 12:27:03 -05:00
Ginger d1e5c17cb2 chore: Formatting 2025-11-13 12:27:03 -05:00
ginger 4f0505ead6 fix: Nitpicky comment reword 2025-11-13 12:27:03 -05:00
Ginger 2f2a233839 fix: Bump max startup time to ten minutes in the systemd unit 2025-11-13 12:27:03 -05:00
Ginger c1dc640da7 chore(sync/v3): More goat sacrifices 2025-11-13 12:27:03 -05:00
Ginger 48fe06a36c refactor(sync/v3): Split load_joined_room into smaller functions 2025-11-13 12:27:03 -05:00
ginger 9150131e37 fix: Correct error message 2025-11-13 12:27:03 -05:00
Ginger 6414a6da9b fix(sync/v3): Add a workaround for matrix-js-sdk/5071 2025-11-13 12:27:03 -05:00
Ginger e576fcd8ee fix(sync/v3): Stop ignoring leave cache deserialization failures 2025-11-13 12:27:03 -05:00
Ginger 6e7710b657 fix(sync/v3): Do not include the last membership event when syncing left rooms 2025-11-13 12:27:03 -05:00
Ginger e60ce4dca7 chore(sync/v3): Sacrifice a goat to clippy 2025-11-13 12:27:03 -05:00
Ginger 39b8ba6593 fix(sync/v3): Cache shortstatehashes to speed up migration 2025-11-13 12:27:03 -05:00
Ginger 8495acf8f8 fix(sync/v3): Implement a migration for the userroomid_leftstate table 2025-11-13 12:27:03 -05:00
Ginger 90bd95e3c7 fix(sync/v3): Fix invite filtering for federated invites 2025-11-13 12:27:03 -05:00
Ginger 2d97b5facf feat(sync/v3): Remove TL size config option in favor of using the sync filter 2025-11-13 12:27:03 -05:00
Ginger ed062e4b86 chore(sync/v3): Fix clippy lints 2025-11-13 12:27:03 -05:00
Ginger ae9f57339c fix(sync/v3): Remove mysterious membership event manipulation code 2025-11-13 12:27:03 -05:00
Ginger be7fb291e5 fix(sync/v3): Properly sync room heroes 2025-11-13 12:27:03 -05:00
Ginger 94224f00b3 chore(sync/v3): Use "build_*" terminology instead of "calculate_*" 2025-11-13 12:27:03 -05:00
Ginger 1704ca0755 chore(sync/v3): Use more descriptive names for SyncContext properties 2025-11-13 12:27:03 -05:00
Jade Ellis fb6dba7a04 ci: Don't force override rust version by default 2025-11-13 12:27:03 -05:00
Ginger 74886a8697 chore: Remove unneeded comment 2025-11-13 12:27:03 -05:00
Ginger e1f558d91a fix: Use prepare_lazily_loaded_members for joined rooms
Also, don't take read receipts into consideration for lazy loading.
Synapse doesn't do this and they're making initial syncs very large.
2025-11-13 12:27:03 -05:00
Ginger 4399fb5e2b chore: Clippy fixes 2025-11-13 12:27:03 -05:00
Jade Ellis 7050da1c64 feat: Typing notifications in simplified sliding sync
What's missing? Being able to use separate rooms & lists for typing
indicators.
At the moment, we use the same ones as we use for the timeline, as
todo_rooms is quite intertwined. We need to disentangle this to get that
functionality, although I'm not sure if clients use it.
2025-11-13 12:27:03 -05:00
Ginger d8a4aab6f5 feat: Add a config option to change the max TL size for legacy sync 2025-11-13 12:27:03 -05:00
Ginger d426c423c7 fix: Set limited to true for newly joined rooms again 2025-11-13 12:27:03 -05:00
Ginger e4870eca8f fix: Properly sync left rooms
- Remove most usages of `update_membership` in favor
  of directly calling the `mark_as_*` functions
- Store the leave membership event as the value in the
  `userroomid_leftstate` table
- Use the `userroomid_leftstate` table to synchronize the
  timeline and state for left rooms if possible
2025-11-13 12:27:03 -05:00
Ginger e422bc3ac3 fix: Properly sync newly joined rooms 2025-11-13 12:27:03 -05:00
Ginger c723cd4f88 fix(sync/v3): Further cleanup + improve incremental sync consistency 2025-11-13 12:27:03 -05:00
Ginger b14b077a60 fix: Correctly send limited timelines again 2025-11-13 12:27:03 -05:00
Ginger c22cf1a0e4 refactor: Split sync v3 into multiple files 2025-11-13 12:27:03 -05:00
Ginger 6cc86f7a87 feat: Drop support for MSC3575 (legacy sliding sync) 2025-11-13 12:27:03 -05:00
Ginger b25a9c1bda chore: Clippy fixes 2025-11-13 12:27:03 -05:00
Ginger 1560283d79 fix(sync/v3): Cleanup part 1: mostly fix redundant data in state 2025-11-13 12:27:03 -05:00
Renovate Bot 2ec771c84d chore(deps): update rust crate bytesize to v2.2.0 2025-11-13 05:03:54 +00:00
timedout 9375e81974 fix(1163): Resolve algorithm misinterpretations 2025-11-13 03:33:47 +00:00
Renovate Bot f22f35d27b chore(deps): update rust crate syn to v2.0.110 2025-11-12 05:03:14 +00:00
Renovate Bot d5c7d80709 chore(deps): update dependency cargo-bins/cargo-binstall to v1.15.11 2025-11-11 23:11:19 +00:00
Jade Ellis 1899d8bb00 ci: Mirror to Docker Hub 2025-11-11 23:11:09 +00:00
Ginger 9a5ba6171f ci: Remove hardcoded default in setup-rust action 2025-11-11 10:37:03 -05:00
renovate da3efa05b5 chore(Nix): Updated flake hashes 2025-11-11 15:07:05 +00:00
Ginger b53ba2eef4 ci: Give flake hashes workflow permissions to push 2025-11-11 15:07:05 +00:00
Jade Ellis 33019c4529 chore: Update rust 2025-11-11 15:07:05 +00:00
Jade Ellis f7bd9eaba8 chore(clippy): Remove old redundant lint 2025-11-11 13:59:12 +00:00
Jade Ellis f9c42bbadc refactor(clippy): Unused self 2025-11-11 13:59:12 +00:00
Jade Ellis fe62c39501 style(clippy): Remove unneeded allocation 2025-11-11 13:59:12 +00:00
Jade Ellis 35320cf0d4 style(clippy): Elide lifetimes 2025-11-11 13:59:12 +00:00
Jade Ellis eaf6a889c2 style(clippy): Unnecessary move
Function is used in a single place and the move doesn't seem to provide
any safety benefits, so 💨
2025-11-11 13:59:12 +00:00
Jade Ellis b04f1332db style(clippy): Remove dead code
Looks like this has been dead since we forked at least, seems pretty
safe to remove
2025-11-11 13:59:12 +00:00
Jade Ellis 9e4bcda17b style(clippy): Make the event graph generic over the hasher 2025-11-11 13:59:12 +00:00
Jade 45e4053883 fix: Don't break when encountering the server user, as there may be real users after 2025-11-10 23:56:02 +00:00
Jade Ellis c0b617f4f1 feat(sentry): Include the commit hash in the release name 2025-11-10 16:57:24 +00:00
Jade Ellis a28cfd284b chore(deps): Upgrade tracing / telemetry ecosystem
We no longer need the tracing patches, so I've removed those and
unpinned them in renovate.

otel's jaeger propagator is deprecated too, so it's replaced with the
builtin W3C TraceContext propagator
2025-11-10 16:42:28 +00:00
Jade Ellis a5b9cb69bd fix(deps): Pin hyper-util back to the patched version 2025-11-10 15:56:09 +00:00
Renovate Bot 3c8f252a14 chore(deps): update opentelemetry-rust monorepo to 0.31.0 2025-11-10 05:03:16 +00:00
Jade 8a63818f31 feat: Enable sentry compilation feature 2025-11-10 01:33:50 +00:00
Renovate Bot 5b5e26e529 chore(deps): update dependency cargo-bins/cargo-binstall to v1.15.10 2025-11-09 19:05:26 +00:00
aviac 866769c054 chore: replace serde-yml with serde-saphyr
- serde-yml has an un-addressed [security issue][sec-issue]
- [saphyr][saphyr] is a pretty recent and active crate that deals with YAML parsing
- based on that, someone recently created [serde-saphyr][serde-saphyr]

---

The change was pretty straightforward and mostly "just a search and replace". The new crate has it's `Error` type split
into serialization and derserialization errors. Hence I created one Continuwuity-Error variant for each instead of just
having a single `Yaml` variant. This was already done previously with the `Toml` errors so I thought this would be
rather acceptable.

[sec-issue]: https://github.com/advisories/GHSA-gfxp-f68g-8x78
[saphyr]: https://github.com/saphyr-rs/saphyr
[serde-saphyr]: https://github.com/saphyr-rs/saphyr/issues/66#issuecomment-3353212289
2025-11-09 11:23:32 +01:00
Renovate Bot 2e3b71f5f1 chore(deps): update rust-patch-updates 2025-11-08 23:57:36 +00:00
Jade 1312d61141 revert f7867cf6ca
revert ci: Clean up old images
2025-11-08 23:56:02 +00:00
Jade Ellis f7867cf6ca ci: Clean up old images 2025-11-08 23:29:25 +00:00
Jade Ellis 2ca6887a5d chore(ci): Fix merge error 2025-11-08 23:08:10 +00:00
Jade Ellis 368685f8cd ci: Re-run mirror script when files change 2025-11-08 23:00:37 +00:00
Jade Ellis ad2d192b94 ci: Use PATs for github registry
https://stackoverflow.com/questions/76821352/how-can-you-authenticate-to-the-github-container-registry-using-a-github-app

thx github
2025-11-08 23:00:31 +00:00
Jade Ellis 3214e94cdb ci: Mirror to ghcr 2025-11-08 22:59:27 +00:00
timedout 37c537379d chore(ci): Add git.nexy7574.co.uk image mirror (#1149)
secrets were added to the org

Reviewed-on: https://forgejo.ellis.link/continuwuation/continuwuity/pulls/1149
Co-authored-by: timedout <git@nexy7574.co.uk>
Co-committed-by: timedout <git@nexy7574.co.uk>
2025-11-08 22:56:16 +00:00
Jade Ellis 3c01c5f085 chore: Don't try to update patched deps automatically 2025-11-08 21:17:04 +00:00
Renovate Bot 4c552bb8ca chore(deps): update pre-commit hook pre-commit/pre-commit-hooks to v6 2025-11-08 20:56:00 +00:00
Jade Ellis ce73d29855 chore: Fix typos 2025-11-08 20:54:49 +00:00
Renovate Bot d6e314744b chore(deps): update pre-commit hook crate-ci/typos to v1.39.0 2025-11-08 14:34:13 +00:00
Jade ec603188de ci: Enable pre-commit in renovate 2025-11-08 14:31:35 +00:00
timedout fbf48addc7 fix(user_can): Fix room creators being unable to redact events in v12 rooms 2025-10-27 14:34:50 +00:00
nexy7574 cbf726580f fix: Kicks in !v12 are impossible 2025-10-27 14:34:50 +00:00
nexy7574 28f258fc8c fix: Incorrect interpretation of 5.5.4 2025-10-27 14:34:50 +00:00
nexy7574 8b3acfd770 fix: Inverted creatorship check 2025-10-27 14:34:50 +00:00
nexy7574 a581e8de01 fix: Don't check restricted join rules for invite joins 2025-10-27 14:34:50 +00:00
nexy7574 7c74db5e74 fix: Weird re-application of partially resolved state 2025-10-27 14:34:50 +00:00
nexy7574 b17b4235f3 fix: Unbans and kicks incorrectly checked creatorship in !v12 2025-10-27 14:34:50 +00:00
aviac ec3564e8aa chore: use upstream rust-jemalloc-sys-unprefixed after flake.lock update 2025-10-27 12:55:21 +00:00
aviac 9a887ac04b chore: fix CI to make all checks green
- define a nix default package
- try to fix CI
- fix/improve (?) CI even more (??)
2025-10-27 12:55:21 +00:00
aviac fed808a3c6 feat: add taplo.toml to check now that we have it 2025-10-27 12:55:21 +00:00
aviac 37983b33a2 feat: add treefmt 2025-10-27 12:55:21 +00:00
aviac 1b2224fac6 feat: add hydra jobs to build all packages 2025-10-27 12:55:21 +00:00
aviac c1c165ab48 fix: apply rocksdb changes in checks and shll 2025-10-27 12:55:20 +00:00
aviac 68bea1816f feat(nix): flake-parts, first draft 2025-10-27 12:55:20 +00:00
Odd E. Ebbesen cb7875e479 fix(#1134): Update docs and implementation of admin media delete-past-remote-media (#1136)
Reviewed-on: https://forgejo.ellis.link/continuwuation/continuwuity/pulls/1136
Co-authored-by: Odd E. Ebbesen <git@oddware.net>
Co-committed-by: Odd E. Ebbesen <git@oddware.net>
2025-10-27 12:31:25 +00:00
Jade Ellis 910a3182f7 fix: Prevent crash on process exit on MacOS 2025-10-26 17:42:08 +00:00
Jade Ellis 05886f8dcb feat: Add option to control WAL compression
Also enables zstd compression by default
2025-10-26 17:30:42 +00:00
timedout cff3c27729 fix: Bump ruwuma, export new route, config loading 2025-10-24 16:37:22 -04:00
Kierre 80be2ca22c Repair 2025-10-24 16:07:06 -04:00
Kierre d133b6c0c3 feat: set MSC4373 values 2025-10-24 15:33:16 -04:00
Ginger a3592bd3b7 feat: Make a few improvements to the systemd unit
- Use systemd's credential system to supply our config file
- Remove `ConfigurationDirectory` to prevent conflicts with package managers
- Set `config_reload_signal` to true using an envvar
2025-10-17 13:37:42 +00:00
Ginger 70e8e96302 fix: Use mode 600 for config files on Fedora because they contain secret info 2025-10-17 13:37:42 +00:00
timedout 6002edccd3 perf: Remove extraneous policy server check 2025-10-16 23:57:07 +01:00
timedout d189004d65 feat: Add more granular controls for policy server calling (#1127)
Adds two new toggles to the configuration, the first of which allows disabling the policy server checks entirely, and the second of which allows disabling checking events created locally. They're both enabled by default for maximum PS efficacy but allowing them to be disabled allows people who frequently cannot contact policy servers, for example those in censored countries, to be able to still use rooms with pace, allows single-user/trusted-only homeservers to disable the preliminary check on their own events, and also gives an escape hatch in case an issue like #1060 happens again, especially with MSCs not in FCP being moving targets.

In future, I think we should gate all MSC implementations behind config flags, even if they default to on.

Reviewed-on: https://forgejo.ellis.link/continuwuation/continuwuity/pulls/1127
Reviewed-by: Jade Ellis <jade@ellis.link>
Co-authored-by: timedout <git@nexy7574.co.uk>
Co-committed-by: timedout <git@nexy7574.co.uk>
2025-10-16 22:45:23 +00:00
timedout 26b700bf51 fix: Policy server calls use the correct JSON object (#1126)
Fixes #1060

Reviewed-on: https://forgejo.ellis.link/continuwuation/continuwuity/pulls/1126
Reviewed-by: Jacob Taylor <aranjedeath@noreply.forgejo.ellis.link>
Co-authored-by: timedout <git@nexy7574.co.uk>
Co-committed-by: timedout <git@nexy7574.co.uk>
2025-10-16 21:06:54 +00:00
Renovate Bot 09f24745c3 chore(deps): lock file maintenance 2025-10-15 19:05:50 +00:00
Renovate Bot 7ffbbe6890 chore(deps): update https://github.com/actions/setup-node action to v6 2025-10-15 18:55:02 +00:00
Renovate Bot ad94c112fe chore(deps): update rust-patch-updates 2025-10-15 17:55:58 +00:00
Jade 8c7cc68cbf fix(ci): Don't use shallow clone when we're comparing git history 2025-10-15 12:53:15 +00:00
Ginger dc047b635f feat: Send notifications to systemd when a reload is triggered 2025-10-15 03:12:25 +00:00
Renovate Bot cc4c2fed25 chore(deps): lock file maintenance 2025-10-13 12:05:52 +00:00
Renovate Bot 17e47ecd6d chore(deps): update github-actions-non-major 2025-10-13 11:27:22 +00:00
Jade b1d5ff477b chore: Update renovate config
- Limit renovate updates to mondays
- Don't group lock updates
- Update checksums if possible
2025-10-13 11:26:26 +00:00
Renovate Bot d6dc01ac2c chore(deps): update https://code.forgejo.org/actions/checkout action to v5 2025-10-13 10:41:20 +00:00
Jimmy Brush 77ebe0d02f fix(!714): Off-by-one in v5 sync
Simplified sliding sync specifies ranges to be inclusive while rust ranges are
exclusive.
2025-10-13 10:28:19 +00:00
Renovate Bot 81e3d4c905 chore(deps): update dependency cargo-bins/cargo-binstall to v1.15.7 2025-10-13 10:27:18 +00:00
nexy7574 cb8f36444c feat: Proactively read Content-Length to reject oversized uploads 2025-10-12 19:42:57 +00:00
nexy7574 799def70dc feat: Produce even more informative errors when saving media fails 2025-10-12 19:42:57 +00:00
nexy7574 20f741d0e5 feat: Produce a more informative error when uploading media fails 2025-10-12 19:42:57 +00:00
Renovate Bot d38f4a24f2 chore(deps): update ghcr.io/renovatebot/renovate docker tag to v41.146.0 2025-10-11 05:03:03 +00:00
Renovate Bot 6604cc4df9 chore(deps): update ghcr.io/renovatebot/renovate docker tag to v41.144.1 2025-10-10 05:01:39 +00:00
Renovate Bot 89aa4d1eae chore(deps): update ghcr.io/renovatebot/renovate docker tag to v41.143.1 2025-10-09 05:03:56 +00:00
Renovate Bot 9231ea5114 chore(deps): update ghcr.io/renovatebot/renovate docker tag to v41.141.0 2025-10-08 05:01:41 +00:00
Renovate Bot 4a3c72338d chore(deps): update ghcr.io/renovatebot/renovate docker tag to v41.138.1 2025-10-07 05:02:54 +00:00
Renovate Bot ab862f4383 chore(deps): update ghcr.io/renovatebot/renovate docker tag to v41.135.5 2025-10-06 05:01:26 +00:00
Renovate Bot bd43be931a chore(deps): update ghcr.io/renovatebot/renovate docker tag to v41.135.4 2025-10-05 05:03:52 +00:00
Ginger 148240cbbb fix: Add missing ldap3 feature 2025-10-01 18:55:30 +00:00
Renovate Bot 2e9e42d9ae chore(deps): update rust crate ldap3 to 0.12.0 2025-10-01 18:55:30 +00:00
Renovate Bot 89fbda0d6e chore(deps): update ghcr.io/renovatebot/renovate docker tag to v41.132.5 2025-10-01 05:03:28 +00:00
Renovate Bot c97eb5c889 chore(deps): update ghcr.io/renovatebot/renovate docker tag to v41.132.2 2025-09-30 05:01:26 +00:00
Ginger 366ec46b26 fix: Upload debs built on a schedule 2025-09-29 14:17:44 +00:00
ginger 62a98ebc71 fix: Upload RPMs built on a schedule 2025-09-29 14:17:44 +00:00
Renovate Bot 439c605efe chore(deps): update ghcr.io/renovatebot/renovate docker tag to v41.131.9 2025-09-29 05:03:13 +00:00
Renovate Bot 32df2f3487 chore(deps): update ghcr.io/renovatebot/renovate docker tag to v41.131.8 2025-09-28 05:03:46 +00:00
Renovate Bot 692da7ffc2 chore(deps): update dependency cargo-bins/cargo-binstall to v1.15.6 2025-09-27 16:17:44 +00:00
Renovate Bot 1082b24b1d chore(deps): update ghcr.io/renovatebot/renovate docker tag to v41.131.6 2025-09-27 05:03:28 +00:00
nexy7574 f45ceedb8a fix(upgrade): Potentially resolve CI clippy errors
I'm not convinced this isn't a rust bug itself,
but CI was complaining about lifetimes
and those complaints couldn't be reproduced locally,
so this should probably fix it maybe?
2025-09-26 18:47:49 +01:00
nexy7574 d614e43981 fix(stateres): Creators can always unban
Also basically rewrote all of the event auth logs to be more digestable
2025-09-26 18:47:49 +01:00
Renovate Bot 1e0e7a31aa chore(deps): update ghcr.io/renovatebot/renovate docker tag to v41.131.2 2025-09-26 05:02:43 +00:00
113 changed files with 4918 additions and 4923 deletions
+68 -62
View File
@@ -17,9 +17,9 @@ inputs:
required: false
default: ''
rust-version:
description: 'Rust version to install (e.g. nightly). Defaults to 1.87.0'
description: 'Rust version to install (e.g. nightly).'
required: false
default: '1.87.0'
default: ''
sccache-cache-limit:
description: 'Maximum size limit for sccache local cache (e.g. 2G, 500M)'
required: false
@@ -59,23 +59,7 @@ runs:
mkdir -p "${{ github.workspace }}/target"
mkdir -p "${{ github.workspace }}/.rustup"
- name: Start cache restore group
shell: bash
run: echo "::group::📦 Restoring caches (registry, toolchain, build artifacts)"
- name: Cache Cargo registry and git
id: registry-cache
uses: actions/cache@v4
with:
path: |
.cargo/registry/index
.cargo/registry/cache
.cargo/git/db
# Registry cache saved per workflow, restored from any workflow's cache
# Each workflow maintains its own registry that accumulates its needed crates
key: continuwuity-cargo-registry-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-${{ github.workflow }}
restore-keys: |
continuwuity-cargo-registry-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-
- name: Cache toolchain binaries
id: toolchain-cache
@@ -88,45 +72,6 @@ runs:
# Shared toolchain cache across all Rust versions
key: continuwuity-toolchain-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}
- name: Setup sccache
uses: https://git.tomfos.tr/tom/sccache-action@v1
- name: Cache dependencies
id: deps-cache
uses: actions/cache@v4
with:
path: |
target/**/.fingerprint
target/**/deps
target/**/*.d
target/**/.cargo-lock
target/**/CACHEDIR.TAG
target/**/.rustc_info.json
/timelord/
# Dependencies cache - based on Cargo.lock, survives source code changes
key: >-
continuwuity-deps-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-${{ inputs.rust-version }}${{ inputs.cache-key-suffix && format('-{0}', inputs.cache-key-suffix) || '' }}-${{ hashFiles('rust-toolchain.toml', '**/Cargo.lock') }}
restore-keys: |
continuwuity-deps-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-${{ inputs.rust-version }}${{ inputs.cache-key-suffix && format('-{0}', inputs.cache-key-suffix) || '' }}-
- name: Cache incremental compilation
id: incremental-cache
uses: actions/cache@v4
with:
path: |
target/**/incremental
# Incremental cache - based on source code changes
key: >-
continuwuity-incremental-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-${{ inputs.rust-version }}${{ inputs.cache-key-suffix && format('-{0}', inputs.cache-key-suffix) || '' }}-${{ hashFiles('rust-toolchain.toml', '**/Cargo.lock') }}-${{ hashFiles('**/*.rs', '**/Cargo.toml') }}
restore-keys: |
continuwuity-incremental-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-${{ inputs.rust-version }}${{ inputs.cache-key-suffix && format('-{0}', inputs.cache-key-suffix) || '' }}-${{ hashFiles('rust-toolchain.toml', '**/Cargo.lock') }}-
continuwuity-incremental-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-${{ inputs.rust-version }}${{ inputs.cache-key-suffix && format('-{0}', inputs.cache-key-suffix) || '' }}-
- name: End cache restore group
shell: bash
run: echo "::endgroup::"
- name: Setup Rust toolchain
shell: bash
run: |
@@ -158,6 +103,71 @@ runs:
fi
echo "::endgroup::"
- name: Output Rust version
id: rust-setup
shell: bash
run: |
RUST_VERSION=$(rustc --version | cut -d' ' -f2)
echo "version=$RUST_VERSION" >> $GITHUB_OUTPUT
echo "📋 Rust version: $(rustc --version)"
- name: Setup sccache
uses: https://git.tomfos.tr/tom/sccache-action@v1
- name: Start cache restore group
shell: bash
run: echo "::group::📦 Restoring caches (registry, toolchain, build artifacts)"
- name: Cache Cargo registry and git
id: registry-cache
uses: actions/cache@v4
with:
path: |
.cargo/registry/index
.cargo/registry/cache
.cargo/git/db
# Registry cache saved per workflow, restored from any workflow's cache
# Each workflow maintains its own registry that accumulates its needed crates
key: continuwuity-cargo-registry-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-${{ github.workflow }}
restore-keys: |
continuwuity-cargo-registry-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-
- name: Cache dependencies
id: deps-cache
uses: actions/cache@v4
with:
path: |
target/**/.fingerprint
target/**/deps
target/**/*.d
target/**/.cargo-lock
target/**/CACHEDIR.TAG
target/**/.rustc_info.json
/timelord/
# Dependencies cache - based on Cargo.lock, survives source code changes
key: >-
continuwuity-deps-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-${{ steps.rust-setup.outputs.version }}${{ inputs.cache-key-suffix && format('-{0}', inputs.cache-key-suffix) || '' }}-${{ hashFiles('rust-toolchain.toml', '**/Cargo.lock') }}
restore-keys: |
continuwuity-deps-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-${{ steps.rust-setup.outputs.version }}${{ inputs.cache-key-suffix && format('-{0}', inputs.cache-key-suffix) || '' }}-
- name: Cache incremental compilation
id: incremental-cache
uses: actions/cache@v4
with:
path: |
target/**/incremental
# Incremental cache - based on source code changes
key: >-
continuwuity-incremental-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-${{ steps.rust-setup.outputs.version }}${{ inputs.cache-key-suffix && format('-{0}', inputs.cache-key-suffix) || '' }}-${{ hashFiles('rust-toolchain.toml', '**/Cargo.lock') }}-${{ hashFiles('**/*.rs', '**/Cargo.toml') }}
restore-keys: |
continuwuity-incremental-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-${{ steps.rust-setup.outputs.version }}${{ inputs.cache-key-suffix && format('-{0}', inputs.cache-key-suffix) || '' }}-${{ hashFiles('rust-toolchain.toml', '**/Cargo.lock') }}-
continuwuity-incremental-${{ steps.runner-os.outputs.slug }}-${{ steps.runner-os.outputs.arch }}-${{ steps.rust-setup.outputs.version }}${{ inputs.cache-key-suffix && format('-{0}', inputs.cache-key-suffix) || '' }}-
- name: End cache restore group
shell: bash
run: echo "::endgroup::"
- name: Configure PATH and install tools
shell: bash
env:
@@ -225,13 +235,9 @@ runs:
echo "📦 Installing target: ${{ inputs.rust-target }}"
rustup target add ${{ inputs.rust-target }}
- name: Output version and summary
id: rust-setup
- name: Output summary
shell: bash
run: |
RUST_VERSION=$(rustc --version | cut -d' ' -f2)
echo "version=$RUST_VERSION" >> $GITHUB_OUTPUT
echo "📋 Setup complete:"
echo " Rust: $(rustc --version)"
echo " Cargo: $(cargo --version)"
+21
View File
@@ -40,6 +40,15 @@ creds:
- registry: registry.gitlab.com
user: "{{env \"GITLAB_USERNAME\"}}"
pass: "{{env \"GITLAB_TOKEN\"}}"
- registry: git.nexy7574.co.uk
user: "{{env \"N7574_GIT_USERNAME\"}}"
pass: "{{env \"N7574_GIT_TOKEN\"}}"
- registry: ghcr.io
user: "{{env \"GH_PACKAGES_USER\"}}"
pass: "{{env \"GH_PACKAGES_TOKEN\"}}"
- registry: docker.io
user: "{{env \"DOCKER_MIRROR_USER\"}}"
pass: "{{env \"DOCKER_MIRROR_TOKEN\"}}"
# Global defaults
defaults:
@@ -53,3 +62,15 @@ sync:
target: registry.gitlab.com/continuwuity/continuwuity
type: repository
<<: *tags-main
- source: *source
target: git.nexy7574.co.uk/mirrored/continuwuity
type: repository
<<: *tags-releases
- source: *source
target: ghcr.io/continuwuity/continuwuity
type: repository
<<: *tags-main
- source: *source
target: docker.io/jadedblueeyes/continuwuity
type: repository
<<: *tags-main
+2 -2
View File
@@ -32,7 +32,7 @@ jobs:
echo "Debian distribution: $DISTRIBUTION ($VERSION)"
- name: Checkout repository with full history
uses: https://code.forgejo.org/actions/checkout@v4
uses: https://code.forgejo.org/actions/checkout@v5
with:
fetch-depth: 0
@@ -132,7 +132,7 @@ jobs:
path: ${{ steps.cargo-deb.outputs.path }}
- name: Publish to Forgejo package registry
if: ${{ forge.event_name == 'push' || forge.event_name == 'workflow_dispatch' }}
if: ${{ forge.event_name == 'push' || forge.event_name == 'workflow_dispatch' || forge.event_name == 'schedule' }}
run: |
OWNER="continuwuation"
DISTRIBUTION=${{ steps.debian-version.outputs.distribution }}
+2 -2
View File
@@ -30,7 +30,7 @@ jobs:
echo "Fedora version: $VERSION"
- name: Checkout repository with full history
uses: https://code.forgejo.org/actions/checkout@v4
uses: https://code.forgejo.org/actions/checkout@v5
with:
fetch-depth: 0
@@ -250,7 +250,7 @@ jobs:
path: artifacts/*debuginfo*.rpm
- name: Publish to RPM Package Registry
if: ${{ github.event_name == 'push' || github.event_name == 'workflow_dispatch' }}
if: ${{ github.event_name == 'push' || github.event_name == 'workflow_dispatch' || github.event_name == 'schedule' }}
run: |
# Find the main binary RPM (exclude debug and source RPMs)
RPM=$(find artifacts -name "continuwuity-*.rpm" \
+1 -1
View File
@@ -55,7 +55,7 @@ jobs:
- name: Setup Node.js
if: steps.runner-env.outputs.node_major == '' || steps.runner-env.outputs.node_major < '20'
uses: https://github.com/actions/setup-node@v5
uses: https://github.com/actions/setup-node@v6
with:
node-version: 22
+1 -1
View File
@@ -24,7 +24,7 @@ jobs:
steps:
- name: 📦 Setup Node.js
uses: https://github.com/actions/setup-node@v5
uses: https://github.com/actions/setup-node@v6
with:
node-version: "22"
+22 -1
View File
@@ -11,7 +11,13 @@ on:
required: false
default: false
type: boolean
push:
branches:
- main
paths:
# Re-run when config changes
- '.forgejo/regsync/regsync.yml'
- '.forgejo/workflows/mirror-images.yml'
concurrency:
group: "mirror-images"
cancel-in-progress: true
@@ -24,12 +30,27 @@ jobs:
BUILTIN_REGISTRY_PASSWORD: ${{ secrets.BUILTIN_REGISTRY_PASSWORD }}
GITLAB_USERNAME: ${{ vars.GITLAB_USERNAME }}
GITLAB_TOKEN: ${{ secrets.GITLAB_TOKEN }}
N7574_GIT_USERNAME: ${{ vars.N7574_GIT_USERNAME }}
N7574_GIT_TOKEN: ${{ secrets.N7574_GIT_TOKEN }}
GH_PACKAGES_USER: ${{ vars.GH_PACKAGES_USER }}
GH_PACKAGES_TOKEN: ${{ secrets.GH_PACKAGES_TOKEN }}
DOCKER_MIRROR_USER: ${{ vars.DOCKER_MIRROR_USER }}
DOCKER_MIRROR_TOKEN: ${{ secrets.DOCKER_MIRROR_TOKEN }}
steps:
- name: Checkout repository
uses: actions/checkout@v5
with:
persist-credentials: false
# - uses: https://github.com/actions/create-github-app-token@v2
# id: app-token
# with:
# app-id: ${{ vars.GH_APP_ID }}
# private-key: ${{ secrets.GH_APP_PRIVATE_KEY }}
# github-api-url: https://api.github.com
# owner: continuwuity
# repositories: continuwuity
- name: Install regctl
uses: https://forgejo.ellis.link/continuwuation/regclient-actions/regctl-installer@main
with:
+1 -1
View File
@@ -43,7 +43,7 @@ jobs:
name: Renovate
runs-on: ubuntu-latest
container:
image: ghcr.io/renovatebot/renovate:41.130.1@sha256:44c4cfceb47d283b5adc654ea445909de4d89e69f94b109e522ca32593a436b5
image: ghcr.io/renovatebot/renovate:41.146.4@sha256:bb70194b7405faf10a6f279b60caa10403a440ba37d158c5a4ef0ae7b67a0f92
options: --tmpfs /tmp:exec
steps:
- name: Checkout
+25 -11
View File
@@ -7,6 +7,8 @@ on:
- "Cargo.lock"
- "Cargo.toml"
- "rust-toolchain.toml"
- "nix/**/*"
- ".forgejo/workflows/update-flake-hashes.yml"
jobs:
update-flake-hashes:
@@ -14,13 +16,14 @@ jobs:
steps:
- uses: https://code.forgejo.org/actions/checkout@08c6903cd8c0fde910a37f88322edcfb5dd907a8 # v5
with:
fetch-depth: 1
fetch-depth: 0
fetch-tags: false
fetch-single-branch: true
submodules: false
persist-credentials: false
persist-credentials: true
token: ${{ secrets.FORGEJO_TOKEN }}
- uses: https://github.com/cachix/install-nix-action@9280e7aca88deada44c930f1e2c78e21c3ae3edd # v31.7.0
- uses: https://github.com/cachix/install-nix-action@7ab6e7fd29da88e74b1e314a4ae9ac6b5cda3801 # v31.8.0
with:
nix_path: nixpkgs=channel:nixos-unstable
@@ -39,13 +42,22 @@ jobs:
echo "Base: $base"
echo "HEAD: $(git rev-parse HEAD)"
git diff --name-only $base HEAD > changed_files.txt
echo "files=$(cat changed_files.txt)" >> $FORGEJO_OUTPUT
echo "detected changes in $(cat changed_files.txt)"
# Join files with commas
files=$(paste -sd, changed_files.txt)
echo "files=$files" >> $FORGEJO_OUTPUT
- name: Debug output
run: |
echo "State of output"
echo "Changed files: ${{ steps.changes.outputs.files }}"
- name: Get new toolchain hash
if: contains(steps.changes.outputs.files, 'Cargo.toml') || contains(steps.changes.outputs.files, 'Cargo.lock') || contains(steps.changes.outputs.files, 'rust-toolchain.toml')
run: |
# Set the current sha256 to an empty hash to make `nix build` calculate a new one
awk '/fromToolchainFile *\{/{found=1; print; next} found && /sha256 =/{sub(/sha256 = .*/, "sha256 = pkgsHost.lib.fakeSha256;"); found=0} 1' flake.nix > temp.nix && mv temp.nix flake.nix
awk '/fromToolchainFile *\{/{found=1; print; next} found && /sha256 =/{sub(/sha256 = .*/, "sha256 = lib.fakeSha256;"); found=0} 1' nix/packages/rust.nix > temp.nix
mv temp.nix nix/packages/rust.nix
# Build continuwuity and filter for the new hash
# We do `|| true` because we want this to fail without stopping the workflow
@@ -53,19 +65,21 @@ jobs:
# Place the new hash in place of the empty hash
new_hash=$(cat new_toolchain_hash.txt)
sed -i "s|pkgsHost.lib.fakeSha256|\"$new_hash\"|" flake.nix
sed -i "s|lib.fakeSha256|\"$new_hash\"|" nix/packages/rust.nix
echo "New hash:"
awk -F'"' '/fromToolchainFile/{found=1; next} found && /sha256 =/{print $2; found=0}' flake.nix
awk -F'"' '/fromToolchainFile/{found=1; next} found && /sha256 =/{print $2; found=0}' nix/packages/rust.nix
echo "Expected new hash:"
cat new_toolchain_hash.txt
rm new_toolchain_hash.txt
- name: Get new rocksdb hash
if: contains(steps.changes.outputs.files, '.nix') || contains(steps.changes.outputs.files, 'flake.lock')
run: |
# Set the current sha256 to an empty hash to make `nix build` calculate a new one
awk '/repo = "rocksdb";/{found=1; print; next} found && /sha256 =/{sub(/sha256 = .*/, "sha256 = pkgsHost.lib.fakeSha256;"); found=0} 1' flake.nix > temp.nix && mv temp.nix flake.nix
awk '/repo = "rocksdb";/{found=1; print; next} found && /sha256 =/{sub(/sha256 = .*/, "sha256 = lib.fakeSha256;"); found=0} 1' nix/packages/rocksdb/package.nix > temp.nix
mv temp.nix nix/packages/rocksdb/package.nix
# Build continuwuity and filter for the new hash
# We do `|| true` because we want this to fail without stopping the workflow
@@ -73,17 +87,17 @@ jobs:
# Place the new hash in place of the empty hash
new_hash=$(cat new_rocksdb_hash.txt)
sed -i "s|pkgsHost.lib.fakeSha256|\"$new_hash\"|" flake.nix
sed -i "s|lib.fakeSha256|\"$new_hash\"|" nix/packages/rocksdb/package.nix
echo "New hash:"
awk -F'"' '/repo = "rocksdb";/{found=1; next} found && /sha256 =/{print $2; found=0}' flake.nix
awk -F'"' '/repo = "rocksdb";/{found=1; next} found && /sha256 =/{print $2; found=0}' nix/packages/rocksdb/package.nix
echo "Expected new hash:"
cat new_rocksdb_hash.txt
rm new_rocksdb_hash.txt
- name: Show diff
run: git diff flake.nix
run: git diff flake.nix nix
- name: Push changes
run: |
+2 -2
View File
@@ -7,7 +7,7 @@ default_stages:
repos:
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v5.0.0
rev: v6.0.0
hooks:
- id: fix-byte-order-marker
- id: check-case-conflict
@@ -23,7 +23,7 @@ repos:
- id: check-added-large-files
- repo: https://github.com/crate-ci/typos
rev: v1.26.0
rev: v1.39.0
hooks:
- id: typos
- id: typos
Generated
+625 -1020
View File
File diff suppressed because it is too large Load Diff
+20 -33
View File
@@ -166,8 +166,8 @@ default-features = false
features = ["raw_value"]
# Used for appservice registration files
[workspace.dependencies.serde_yml]
version = "0.0.12"
[workspace.dependencies.serde-saphyr]
version = "0.0.7"
# Used to load forbidden room/user regex from config
[workspace.dependencies.serde_regex]
@@ -210,13 +210,13 @@ default-features = false
version = "0.1.41"
default-features = false
[workspace.dependencies.tracing-subscriber]
version = "0.3.19"
version = "0.3.20"
default-features = false
features = ["env-filter", "std", "tracing", "tracing-log", "ansi", "fmt"]
[workspace.dependencies.tracing-journald]
version = "0.3.1"
[workspace.dependencies.tracing-core]
version = "0.1.33"
version = "0.1.34"
default-features = false
# for URL previews
@@ -286,7 +286,7 @@ features = [
]
[workspace.dependencies.hyper-util]
version = "0.1.11"
version = "=0.1.17"
default-features = false
features = [
"server-auto",
@@ -351,7 +351,7 @@ version = "0.1.2"
# Used for matrix spec type definitions and helpers
[workspace.dependencies.ruma]
git = "https://forgejo.ellis.link/continuwuation/ruwuma"
rev = "d18823471ab3c09e77ff03eea346d4c07e572654"
rev = "b96a99ac85264aa6bf102c12e201090c6474c99d"
features = [
"compat",
"rand",
@@ -412,28 +412,27 @@ default-features = false
# optional opentelemetry, performance measurements, flamegraphs, etc for performance measurements and monitoring
[workspace.dependencies.opentelemetry]
version = "0.30.0"
version = "0.31.0"
[workspace.dependencies.tracing-flame]
version = "0.2.0"
[workspace.dependencies.tracing-opentelemetry]
version = "0.31.0"
version = "0.32.0"
[workspace.dependencies.opentelemetry_sdk]
version = "0.30.0"
version = "0.31.0"
features = ["rt-tokio"]
[workspace.dependencies.opentelemetry-otlp]
version = "0.30.0"
version = "0.31.0"
features = ["http", "trace", "logs", "metrics"]
[workspace.dependencies.opentelemetry-jaeger-propagator]
version = "0.30.0"
# optional sentry metrics for crash/panic reporting
[workspace.dependencies.sentry]
version = "0.42.0"
version = "0.45.0"
default-features = false
features = [
"backtrace",
@@ -449,9 +448,9 @@ features = [
]
[workspace.dependencies.sentry-tracing]
version = "0.42.0"
version = "0.45.0"
[workspace.dependencies.sentry-tower]
version = "0.42.0"
version = "0.45.0"
# jemalloc usage
[workspace.dependencies.tikv-jemalloc-sys]
@@ -477,7 +476,7 @@ default-features = false
features = ["use_std"]
[workspace.dependencies.console-subscriber]
version = "0.4"
version = "0.5"
[workspace.dependencies.nix]
version = "0.30.1"
@@ -551,9 +550,9 @@ features = ["std"]
version = "1.0.2"
[workspace.dependencies.ldap3]
version = "0.11.5"
version = "0.12.0"
default-features = false
features = ["sync", "tls-rustls"]
features = ["sync", "tls-rustls", "rustls-provider"]
[workspace.dependencies.resolv-conf]
version = "0.7.5"
@@ -564,19 +563,7 @@ version = "0.7.5"
# backport of [https://github.com/tokio-rs/tracing/pull/2956] to the 0.1.x branch of tracing.
# we can switch back to upstream if #2956 is merged and backported in the upstream repo.
# https://forgejo.ellis.link/continuwuation/tracing/commit/b348dca742af641c47bc390261f60711c2af573c
[patch.crates-io.tracing-subscriber]
git = "https://forgejo.ellis.link/continuwuation/tracing"
rev = "1e64095a8051a1adf0d1faa307f9f030889ec2aa"
[patch.crates-io.tracing]
git = "https://forgejo.ellis.link/continuwuation/tracing"
rev = "1e64095a8051a1adf0d1faa307f9f030889ec2aa"
[patch.crates-io.tracing-core]
git = "https://forgejo.ellis.link/continuwuation/tracing"
rev = "1e64095a8051a1adf0d1faa307f9f030889ec2aa"
[patch.crates-io.tracing-log]
git = "https://forgejo.ellis.link/continuwuation/tracing"
rev = "1e64095a8051a1adf0d1faa307f9f030889ec2aa"
# adds a tab completion callback: https://forgejo.ellis.link/continuwuation/rustyline-async/src/branch/main/.patchy/0002-add-tab-completion-callback.patch
# adds event for CTRL+\: https://forgejo.ellis.link/continuwuation/rustyline-async/src/branch/main/.patchy/0001-add-event-for-ctrl.patch
@@ -600,7 +587,7 @@ rev = "9c8e51510c35077df888ee72a36b4b05637147da"
# reverts hyperium#148 conflicting with our delicate federation resolver hooks
[patch.crates-io.hyper-util]
git = "https://forgejo.ellis.link/continuwuation/hyper-util"
rev = "e4ae7628fe4fcdacef9788c4c8415317a4489941"
rev = "5886d5292bf704c246206ad72d010d674a7b77d0"
#
# Our crates
@@ -960,7 +947,7 @@ semicolon_outside_block = "warn"
str_to_string = "warn"
string_lit_chars_any = "warn"
string_slice = "warn"
string_to_string = "warn"
suspicious_xor_used_as_pow = "warn"
tests_outside_test_module = "warn"
try_err = "warn"
+28
View File
@@ -957,6 +957,21 @@
#
#rocksdb_bottommost_compression = true
# Compression algorithm for RocksDB's Write-Ahead-Log (WAL).
#
# At present, only ZSTD compression is supported by RocksDB for WAL
# compression. Enabling this can reduce WAL size at the expense of some
# CPU usage during writes.
#
# The options are:
# - "none" = No compression
# - "zstd" = ZSTD compression
#
# For more information on WAL compression, see:
# https://github.com/facebook/rocksdb/wiki/WAL-Compression
#
#rocksdb_wal_compression = "zstd"
# Database recovery mode (for RocksDB WAL corruption).
#
# Use this option when the server reports corruption and refuses to start.
@@ -1497,6 +1512,19 @@
#
#block_non_admin_invites = false
# Enable or disable making requests to MSC4284 Policy Servers.
# It is recommended you keep this enabled unless you experience frequent
# connectivity issues, such as in a restricted networking environment.
#
#enable_msc4284_policy_servers = true
# Enable running locally generated events through configured MSC4284
# policy servers. You may wish to disable this if your server is
# single-user for a slight speed benefit in some rooms, but otherwise
# should leave it enabled.
#
#policy_server_check_own_events = true
# Allow admins to enter commands in rooms other than "#admins" (admin
# room) by prefixing your message with "\!admin" or "\\!admin" followed up
# a normal continuwuity admin command. The reply will be publicly visible
+1 -1
View File
@@ -48,7 +48,7 @@ EOF
# Developer tool versions
# renovate: datasource=github-releases depName=cargo-bins/cargo-binstall
ENV BINSTALL_VERSION=1.15.5
ENV BINSTALL_VERSION=1.15.11
# renovate: datasource=github-releases depName=psastras/sbom-rs
ENV CARGO_SBOM_VERSION=0.9.1
# renovate: datasource=crate depName=lddtree
+1 -1
View File
@@ -18,7 +18,7 @@ RUN --mount=type=cache,target=/etc/apk/cache apk add \
# Developer tool versions
# renovate: datasource=github-releases depName=cargo-bins/cargo-binstall
ENV BINSTALL_VERSION=1.15.5
ENV BINSTALL_VERSION=1.15.11
# renovate: datasource=github-releases depName=psastras/sbom-rs
ENV CARGO_SBOM_VERSION=0.9.1
# renovate: datasource=crate depName=lddtree
+18 -3
View File
@@ -1078,7 +1078,10 @@ ###### **Subcommands:**
* `delete` — - Deletes a single media file from our database and on the filesystem via a single MXC URL or event ID (not redacted)
* `delete-list` — - Deletes a codeblock list of MXC URLs from our database and on the filesystem. This will always ignore errors
* `delete-past-remote-media` - Deletes all remote (and optionally local) media created before or after [duration] time using filesystem metadata first created at date, or fallback to last modified date. This will always ignore errors by default
* `delete-past-remote-media` — Deletes all remote (and optionally local) media created before/after
[duration] ago, using filesystem metadata first created at date, or
fallback to last modified date. This will always ignore errors by
default.
* `delete-all-from-user` — - Deletes all the local media from a local user on our server. This will always ignore errors by default
* `delete-all-from-server` — - Deletes all remote media from the specified remote server. This will always ignore errors by default
* `get-file-info`
@@ -1110,13 +1113,25 @@ ## `admin media delete-list`
## `admin media delete-past-remote-media`
- Deletes all remote (and optionally local) media created before or after [duration] time using filesystem metadata first created at date, or fallback to last modified date. This will always ignore errors by default
Deletes all remote (and optionally local) media created before/after
[duration] ago, using filesystem metadata first created at date, or
fallback to last modified date. This will always ignore errors by
default.
* Examples:
* Delete all remote media older than a year:
`!admin media delete-past-remote-media -b 1y`
* Delete all remote and local media from 3 days ago, up until now:
`!admin media delete-past-remote-media -a 3d --yes-i-want-to-delete-local-media`
**Usage:** `admin media delete-past-remote-media [OPTIONS] <DURATION>`
###### **Arguments:**
* `<DURATION>` — - The relative time (e.g. 30s, 5m, 7d) within which to search
* `<DURATION>` — - The relative time (e.g. 30s, 5m, 7d) from now within which to search
###### **Options:**
+1 -1
View File
@@ -241,7 +241,7 @@ ## Documentation
### Code Comments
- Reference related documentation or parts of the specification
- When a task has multiple ways of being acheved, explain your reasoning for your decision
- When a task has multiple ways of being achieved, explain your reasoning for your decision
- Update comments when code changes
```rs
Generated
+52 -399
View File
@@ -1,94 +1,28 @@
{
"nodes": {
"attic": {
"inputs": {
"crane": "crane",
"flake-compat": "flake-compat",
"flake-parts": "flake-parts",
"nix-github-actions": "nix-github-actions",
"nixpkgs": "nixpkgs",
"nixpkgs-stable": "nixpkgs-stable"
},
"advisory-db": {
"flake": false,
"locked": {
"lastModified": 1757683818,
"narHash": "sha256-q7q0pWT+wu5AUU1Qlbwq8Mqb+AzHKhaMCVUq/HNZfo8=",
"owner": "zhaofengli",
"repo": "attic",
"rev": "7c5d79ad62cda340cb8c80c99b921b7b7ffacf69",
"lastModified": 1761112158,
"narHash": "sha256-RIXu/7eyKpQHjsPuAUODO81I4ni8f+WYSb7K4mTG6+0=",
"owner": "rustsec",
"repo": "advisory-db",
"rev": "58f3aaec0e1776f4a900737be8cd7cb00972210d",
"type": "github"
},
"original": {
"owner": "zhaofengli",
"ref": "main",
"repo": "attic",
"type": "github"
}
},
"cachix": {
"inputs": {
"devenv": "devenv",
"flake-compat": "flake-compat_2",
"git-hooks": "git-hooks",
"nixpkgs": "nixpkgs_2"
},
"locked": {
"lastModified": 1756385612,
"narHash": "sha256-+NU5MMhuPHHRyvZZWNFG7zt+leRSPsJu1MwhOUzkPUk=",
"owner": "cachix",
"repo": "cachix",
"rev": "dc24688cd67518c3711d511fa369c0f5a131063a",
"type": "github"
},
"original": {
"owner": "cachix",
"ref": "master",
"repo": "cachix",
"type": "github"
}
},
"cachix_2": {
"inputs": {
"devenv": [
"cachix",
"devenv"
],
"flake-compat": [
"cachix",
"devenv"
],
"git-hooks": [
"cachix",
"devenv",
"git-hooks"
],
"nixpkgs": [
"cachix",
"devenv",
"nixpkgs"
]
},
"locked": {
"lastModified": 1748883665,
"narHash": "sha256-R0W7uAg+BLoHjMRMQ8+oiSbTq8nkGz5RDpQ+ZfxxP3A=",
"owner": "cachix",
"repo": "cachix",
"rev": "f707778d902af4d62d8dd92c269f8e70de09acbe",
"type": "github"
},
"original": {
"owner": "cachix",
"ref": "latest",
"repo": "cachix",
"owner": "rustsec",
"repo": "advisory-db",
"type": "github"
}
},
"crane": {
"locked": {
"lastModified": 1751562746,
"narHash": "sha256-smpugNIkmDeicNz301Ll1bD7nFOty97T79m4GUMUczA=",
"lastModified": 1760924934,
"narHash": "sha256-tuuqY5aU7cUkR71sO2TraVKK2boYrdW3gCSXUkF4i44=",
"owner": "ipetkov",
"repo": "crane",
"rev": "aed2020fd3dc26e1e857d4107a5a67a33ab6c1fd",
"rev": "c6b4d5308293d0d04fcfeee92705017537cad02f",
"type": "github"
},
"original": {
@@ -97,53 +31,6 @@
"type": "github"
}
},
"crane_2": {
"locked": {
"lastModified": 1757183466,
"narHash": "sha256-kTdCCMuRE+/HNHES5JYsbRHmgtr+l9mOtf5dpcMppVc=",
"owner": "ipetkov",
"repo": "crane",
"rev": "d599ae4847e7f87603e7082d73ca673aa93c916d",
"type": "github"
},
"original": {
"owner": "ipetkov",
"ref": "master",
"repo": "crane",
"type": "github"
}
},
"devenv": {
"inputs": {
"cachix": "cachix_2",
"flake-compat": [
"cachix",
"flake-compat"
],
"git-hooks": [
"cachix",
"git-hooks"
],
"nix": "nix",
"nixpkgs": [
"cachix",
"nixpkgs"
]
},
"locked": {
"lastModified": 1754404745,
"narHash": "sha256-BdbW/iTImczgcuATgQIa9sPGuYIBxVq2xqcvICsa2AQ=",
"owner": "cachix",
"repo": "devenv",
"rev": "6563b21105168f90394dfaf58284b078af2d7275",
"type": "github"
},
"original": {
"owner": "cachix",
"repo": "devenv",
"type": "github"
}
},
"fenix": {
"inputs": {
"nixpkgs": [
@@ -152,53 +39,20 @@
"rust-analyzer-src": "rust-analyzer-src"
},
"locked": {
"lastModified": 1758004879,
"narHash": "sha256-kV7tQzcNbmo58wg2uE2MQ/etaTx+PxBMHeNrLP8vOgk=",
"lastModified": 1761115517,
"narHash": "sha256-Fev/ag/c3Fp3JBwHfup3lpA5FlNXfkoshnQ7dssBgJ0=",
"owner": "nix-community",
"repo": "fenix",
"rev": "07e5ce53dd020e6b337fdddc934561bee0698fa2",
"rev": "320433651636186ea32b387cff05d6bbfa30cea7",
"type": "github"
},
"original": {
"owner": "nix-community",
"ref": "main",
"repo": "fenix",
"type": "github"
}
},
"flake-compat": {
"flake": false,
"locked": {
"lastModified": 1747046372,
"narHash": "sha256-CIVLLkVgvHYbgI2UpXvIIBJ12HWgX+fjA8Xf8PUmqCY=",
"owner": "edolstra",
"repo": "flake-compat",
"rev": "9100a0f413b0c601e0533d1d94ffd501ce2e7885",
"type": "github"
},
"original": {
"owner": "edolstra",
"repo": "flake-compat",
"type": "github"
}
},
"flake-compat_2": {
"flake": false,
"locked": {
"lastModified": 1747046372,
"narHash": "sha256-CIVLLkVgvHYbgI2UpXvIIBJ12HWgX+fjA8Xf8PUmqCY=",
"owner": "edolstra",
"repo": "flake-compat",
"rev": "9100a0f413b0c601e0533d1d94ffd501ce2e7885",
"type": "github"
},
"original": {
"owner": "edolstra",
"repo": "flake-compat",
"type": "github"
}
},
"flake-compat_3": {
"flake": false,
"locked": {
"lastModified": 1747046372,
@@ -217,17 +71,14 @@
},
"flake-parts": {
"inputs": {
"nixpkgs-lib": [
"attic",
"nixpkgs"
]
"nixpkgs-lib": "nixpkgs-lib"
},
"locked": {
"lastModified": 1751413152,
"narHash": "sha256-Tyw1RjYEsp5scoigs1384gIg6e0GoBVjms4aXFfRssQ=",
"lastModified": 1760948891,
"narHash": "sha256-TmWcdiUUaWk8J4lpjzu4gCGxWY6/Ok7mOK4fIFfBuU4=",
"owner": "hercules-ci",
"repo": "flake-parts",
"rev": "77826244401ea9de6e3bac47c2db46005e1f30b5",
"rev": "864599284fc7c0ba6357ed89ed5e2cd5040f0c04",
"type": "github"
},
"original": {
@@ -236,214 +87,13 @@
"type": "github"
}
},
"flake-parts_2": {
"inputs": {
"nixpkgs-lib": [
"cachix",
"devenv",
"nix",
"nixpkgs"
]
},
"locked": {
"lastModified": 1733312601,
"narHash": "sha256-4pDvzqnegAfRkPwO3wmwBhVi/Sye1mzps0zHWYnP88c=",
"owner": "hercules-ci",
"repo": "flake-parts",
"rev": "205b12d8b7cd4802fbcb8e8ef6a0f1408781a4f9",
"type": "github"
},
"original": {
"owner": "hercules-ci",
"repo": "flake-parts",
"type": "github"
}
},
"flake-utils": {
"inputs": {
"systems": "systems"
},
"locked": {
"lastModified": 1731533236,
"narHash": "sha256-l0KFg5HjrsfsO/JpG+r7fRrqm12kzFHyUHqHCVpMMbI=",
"owner": "numtide",
"repo": "flake-utils",
"rev": "11707dc2f618dd54ca8739b309ec4fc024de578b",
"type": "github"
},
"original": {
"owner": "numtide",
"ref": "main",
"repo": "flake-utils",
"type": "github"
}
},
"git-hooks": {
"inputs": {
"flake-compat": [
"cachix",
"flake-compat"
],
"gitignore": "gitignore",
"nixpkgs": [
"cachix",
"nixpkgs"
]
},
"locked": {
"lastModified": 1750779888,
"narHash": "sha256-wibppH3g/E2lxU43ZQHC5yA/7kIKLGxVEnsnVK1BtRg=",
"owner": "cachix",
"repo": "git-hooks.nix",
"rev": "16ec914f6fb6f599ce988427d9d94efddf25fe6d",
"type": "github"
},
"original": {
"owner": "cachix",
"repo": "git-hooks.nix",
"type": "github"
}
},
"gitignore": {
"inputs": {
"nixpkgs": [
"cachix",
"git-hooks",
"nixpkgs"
]
},
"locked": {
"lastModified": 1709087332,
"narHash": "sha256-HG2cCnktfHsKV0s4XW83gU3F57gaTljL9KNSuG6bnQs=",
"owner": "hercules-ci",
"repo": "gitignore.nix",
"rev": "637db329424fd7e46cf4185293b9cc8c88c95394",
"type": "github"
},
"original": {
"owner": "hercules-ci",
"repo": "gitignore.nix",
"type": "github"
}
},
"nix": {
"inputs": {
"flake-compat": [
"cachix",
"devenv",
"flake-compat"
],
"flake-parts": "flake-parts_2",
"git-hooks-nix": [
"cachix",
"devenv",
"git-hooks"
],
"nixpkgs": [
"cachix",
"devenv",
"nixpkgs"
],
"nixpkgs-23-11": [
"cachix",
"devenv"
],
"nixpkgs-regression": [
"cachix",
"devenv"
]
},
"locked": {
"lastModified": 1752773918,
"narHash": "sha256-dOi/M6yNeuJlj88exI+7k154z+hAhFcuB8tZktiW7rg=",
"owner": "cachix",
"repo": "nix",
"rev": "031c3cf42d2e9391eee373507d8c12e0f9606779",
"type": "github"
},
"original": {
"owner": "cachix",
"ref": "devenv-2.30",
"repo": "nix",
"type": "github"
}
},
"nix-filter": {
"locked": {
"lastModified": 1757882181,
"narHash": "sha256-+cCxYIh2UNalTz364p+QYmWHs0P+6wDhiWR4jDIKQIU=",
"owner": "numtide",
"repo": "nix-filter",
"rev": "59c44d1909c72441144b93cf0f054be7fe764de5",
"type": "github"
},
"original": {
"owner": "numtide",
"ref": "main",
"repo": "nix-filter",
"type": "github"
}
},
"nix-github-actions": {
"inputs": {
"nixpkgs": [
"attic",
"nixpkgs"
]
},
"locked": {
"lastModified": 1737420293,
"narHash": "sha256-F1G5ifvqTpJq7fdkT34e/Jy9VCyzd5XfJ9TO8fHhJWE=",
"owner": "nix-community",
"repo": "nix-github-actions",
"rev": "f4158fa080ef4503c8f4c820967d946c2af31ec9",
"type": "github"
},
"original": {
"owner": "nix-community",
"repo": "nix-github-actions",
"type": "github"
}
},
"nixpkgs": {
"locked": {
"lastModified": 1751949589,
"narHash": "sha256-mgFxAPLWw0Kq+C8P3dRrZrOYEQXOtKuYVlo9xvPntt8=",
"lastModified": 1760878510,
"narHash": "sha256-K5Osef2qexezUfs0alLvZ7nQFTGS9DL2oTVsIXsqLgs=",
"owner": "NixOS",
"repo": "nixpkgs",
"rev": "9b008d60392981ad674e04016d25619281550a9d",
"type": "github"
},
"original": {
"owner": "NixOS",
"ref": "nixpkgs-unstable",
"repo": "nixpkgs",
"type": "github"
}
},
"nixpkgs-stable": {
"locked": {
"lastModified": 1751741127,
"narHash": "sha256-t75Shs76NgxjZSgvvZZ9qOmz5zuBE8buUaYD28BMTxg=",
"owner": "NixOS",
"repo": "nixpkgs",
"rev": "29e290002bfff26af1db6f64d070698019460302",
"type": "github"
},
"original": {
"owner": "NixOS",
"ref": "nixos-25.05",
"repo": "nixpkgs",
"type": "github"
}
},
"nixpkgs_2": {
"locked": {
"lastModified": 1754214453,
"narHash": "sha256-Q/I2xJn/j1wpkGhWkQnm20nShYnG7TI99foDBpXm1SY=",
"owner": "NixOS",
"repo": "nixpkgs",
"rev": "5b09dc45f24cf32316283e62aec81ffee3c3e376",
"rev": "5e2a59a5b1a82f89f2c7e598302a9cacebb72a67",
"type": "github"
},
"original": {
@@ -453,42 +103,40 @@
"type": "github"
}
},
"nixpkgs_3": {
"nixpkgs-lib": {
"locked": {
"lastModified": 1758029226,
"narHash": "sha256-TjqVmbpoCqWywY9xIZLTf6ANFvDCXdctCjoYuYPYdMI=",
"owner": "NixOS",
"repo": "nixpkgs",
"rev": "08b8f92ac6354983f5382124fef6006cade4a1c1",
"lastModified": 1754788789,
"narHash": "sha256-x2rJ+Ovzq0sCMpgfgGaaqgBSwY+LST+WbZ6TytnT9Rk=",
"owner": "nix-community",
"repo": "nixpkgs.lib",
"rev": "a73b9c743612e4244d865a2fdee11865283c04e6",
"type": "github"
},
"original": {
"owner": "NixOS",
"ref": "nixpkgs-unstable",
"repo": "nixpkgs",
"owner": "nix-community",
"repo": "nixpkgs.lib",
"type": "github"
}
},
"root": {
"inputs": {
"attic": "attic",
"cachix": "cachix",
"crane": "crane_2",
"advisory-db": "advisory-db",
"crane": "crane",
"fenix": "fenix",
"flake-compat": "flake-compat_3",
"flake-utils": "flake-utils",
"nix-filter": "nix-filter",
"nixpkgs": "nixpkgs_3"
"flake-compat": "flake-compat",
"flake-parts": "flake-parts",
"nixpkgs": "nixpkgs",
"treefmt-nix": "treefmt-nix"
}
},
"rust-analyzer-src": {
"flake": false,
"locked": {
"lastModified": 1757362324,
"narHash": "sha256-/PAhxheUq4WBrW5i/JHzcCqK5fGWwLKdH6/Lu1tyS18=",
"lastModified": 1761077270,
"narHash": "sha256-O1uTuvI/rUlubJ8AXKyzh1WSWV3qCZX0huTFUvWLN4E=",
"owner": "rust-lang",
"repo": "rust-analyzer",
"rev": "9edc9cbe5d8e832b5864e09854fa94861697d2fd",
"rev": "39990a923c8bca38f5bd29dc4c96e20ee7808d5d",
"type": "github"
},
"original": {
@@ -498,18 +146,23 @@
"type": "github"
}
},
"systems": {
"treefmt-nix": {
"inputs": {
"nixpkgs": [
"nixpkgs"
]
},
"locked": {
"lastModified": 1681028828,
"narHash": "sha256-Vy1rq5AaRuLzOxct8nz4T6wlgyUR7zLU309k9mBC768=",
"owner": "nix-systems",
"repo": "default",
"rev": "da67096a3b9bf56a91d16901293e51ba5b49a27e",
"lastModified": 1760945191,
"narHash": "sha256-ZRVs8UqikBa4Ki3X4KCnMBtBW0ux1DaT35tgsnB1jM4=",
"owner": "numtide",
"repo": "treefmt-nix",
"rev": "f56b1934f5f8fcab8deb5d38d42fd692632b47c2",
"type": "github"
},
"original": {
"owner": "nix-systems",
"repo": "default",
"owner": "numtide",
"repo": "treefmt-nix",
"type": "github"
}
}
+33 -338
View File
@@ -1,351 +1,46 @@
{
description = "A nix flake for the continuwuity project";
inputs = {
attic.url = "github:zhaofengli/attic?ref=main";
cachix.url = "github:cachix/cachix?ref=master";
crane = {
url = "github:ipetkov/crane?ref=master";
};
# basics
flake-parts.url = "github:hercules-ci/flake-parts";
nixpkgs.url = "github:NixOS/nixpkgs/nixos-unstable";
# for rust via nix
crane.url = "github:ipetkov/crane";
fenix = {
url = "github:nix-community/fenix?ref=main";
url = "github:nix-community/fenix";
inputs.nixpkgs.follows = "nixpkgs";
};
# for vuln checks
advisory-db = {
url = "github:rustsec/advisory-db";
flake = false;
};
treefmt-nix = {
url = "github:numtide/treefmt-nix";
inputs.nixpkgs.follows = "nixpkgs";
};
# for default.nix
flake-compat = {
url = "github:edolstra/flake-compat?ref=master";
flake = false;
};
flake-utils.url = "github:numtide/flake-utils?ref=main";
nix-filter.url = "github:numtide/nix-filter?ref=main";
nixpkgs.url = "github:NixOS/nixpkgs?ref=nixpkgs-unstable";
};
outputs =
inputs:
inputs.flake-utils.lib.eachDefaultSystem (
system:
let
pkgsHost = import inputs.nixpkgs {
inherit system;
};
fnx = inputs.fenix.packages.${system};
# The Rust toolchain to use
toolchain = fnx.combine [
(fnx.fromToolchainFile {
file = ./rust-toolchain.toml;
# See also `rust-toolchain.toml`
sha256 = "sha256-+9FmLhAOezBZCOziO0Qct1NOrfpjNsXxc/8I0c7BdKE=";
})
fnx.complete.rustfmt
];
mkScope =
pkgs:
pkgs.lib.makeScope pkgs.newScope (self: {
inherit pkgs inputs;
craneLib = (inputs.crane.mkLib pkgs).overrideToolchain (_: toolchain);
main = self.callPackage ./pkg/nix/pkgs/main { };
liburing = pkgs.liburing.overrideAttrs {
# Tests weren't building
outputs = [
"out"
"dev"
"man"
];
buildFlags = [ "library" ];
};
rocksdb =
(pkgs.rocksdb_9_10.override {
# Override the liburing input for the build with our own so
# we have it built with the library flag
inherit (self) liburing;
}).overrideAttrs
(old: {
src = pkgsHost.fetchFromGitea {
domain = "forgejo.ellis.link";
owner = "continuwuation";
repo = "rocksdb";
rev = "10.5.fb";
sha256 = "sha256-X4ApGLkHF9ceBtBg77dimEpu720I79ffLoyPa8JMHaU=";
};
version = "v10.5.fb";
cmakeFlags =
pkgs.lib.subtractLists [
# No real reason to have snappy or zlib, no one uses this
"-DWITH_SNAPPY=1"
"-DZLIB=1"
"-DWITH_ZLIB=1"
# We don't need to use ldb or sst_dump (core_tools)
"-DWITH_CORE_TOOLS=1"
# We don't need to build rocksdb tests
"-DWITH_TESTS=1"
# We use rust-rocksdb via C interface and don't need C++ RTTI
"-DUSE_RTTI=1"
# This doesn't exist in RocksDB, and USE_SSE is deprecated for
# PORTABLE=$(march)
"-DFORCE_SSE42=1"
# PORTABLE will get set in main/default.nix
"-DPORTABLE=1"
] old.cmakeFlags
++ [
# No real reason to have snappy, no one uses this
"-DWITH_SNAPPY=0"
"-DZLIB=0"
"-DWITH_ZLIB=0"
# We don't need to use ldb or sst_dump (core_tools)
"-DWITH_CORE_TOOLS=0"
# We don't need trace tools
"-DWITH_TRACE_TOOLS=0"
# We don't need to build rocksdb tests
"-DWITH_TESTS=0"
# We use rust-rocksdb via C interface and don't need C++ RTTI
"-DUSE_RTTI=0"
];
# outputs has "tools" which we don't need or use
outputs = [ "out" ];
# preInstall hooks has stuff for messing with ldb/sst_dump which we don't need or use
preInstall = "";
# We have this already at https://forgejo.ellis.link/continuwuation/rocksdb/commit/a935c0273e1ba44eacf88ce3685a9b9831486155
# Unsetting this so we don't have to revert it and make this nix exclusive
patches = [ ];
postPatch = ''
# Fix gcc-13 build failures due to missing <cstdint> and
# <system_error> includes, fixed upstream since 8.x
sed -e '1i #include <cstdint>' -i db/compaction/compaction_iteration_stats.h
sed -e '1i #include <cstdint>' -i table/block_based/data_block_hash_index.h
sed -e '1i #include <cstdint>' -i util/string_util.h
sed -e '1i #include <cstdint>' -i include/rocksdb/utilities/checkpoint.h
'';
});
});
scopeHost = mkScope pkgsHost;
mkCrossScope =
crossSystem:
let
pkgsCrossStatic =
(import inputs.nixpkgs {
inherit system;
crossSystem = {
config = crossSystem;
};
}).pkgsStatic;
in
mkScope pkgsCrossStatic;
in
{
packages =
{
default = scopeHost.main.override {
disable_features = [
# Don't include experimental features
"experimental"
# jemalloc profiling/stats features are expensive and shouldn't
# be expected on non-debug builds.
"jemalloc_prof"
"jemalloc_stats"
# This is non-functional on nix for some reason
"hardened_malloc"
# conduwuit_mods is a development-only hot reload feature
"conduwuit_mods"
];
};
default-debug = scopeHost.main.override {
profile = "dev";
# Debug build users expect full logs
disable_release_max_log_level = true;
disable_features = [
# Don't include experimental features
"experimental"
# This is non-functional on nix for some reason
"hardened_malloc"
# conduwuit_mods is a development-only hot reload feature
"conduwuit_mods"
];
};
# Just a test profile used for things like CI and complement
default-test = scopeHost.main.override {
profile = "test";
disable_release_max_log_level = true;
disable_features = [
# Don't include experimental features
"experimental"
# this is non-functional on nix for some reason
"hardened_malloc"
# conduwuit_mods is a development-only hot reload feature
"conduwuit_mods"
];
};
all-features = scopeHost.main.override {
all_features = true;
disable_features = [
# Don't include experimental features
"experimental"
# jemalloc profiling/stats features are expensive and shouldn't
# be expected on non-debug builds.
"jemalloc_prof"
"jemalloc_stats"
# This is non-functional on nix for some reason
"hardened_malloc"
# conduwuit_mods is a development-only hot reload feature
"conduwuit_mods"
];
};
all-features-debug = scopeHost.main.override {
profile = "dev";
all_features = true;
# Debug build users expect full logs
disable_release_max_log_level = true;
disable_features = [
# Don't include experimental features
"experimental"
# This is non-functional on nix for some reason
"hardened_malloc"
# conduwuit_mods is a development-only hot reload feature
"conduwuit_mods"
];
};
hmalloc = scopeHost.main.override { features = [ "hardened_malloc" ]; };
}
// builtins.listToAttrs (
builtins.concatLists (
builtins.map
(
crossSystem:
let
binaryName = "static-${crossSystem}";
scopeCrossStatic = mkCrossScope crossSystem;
in
[
# An output for a statically-linked binary
{
name = binaryName;
value = scopeCrossStatic.main;
}
# An output for a statically-linked binary with x86_64 haswell
# target optimisations
{
name = "${binaryName}-x86_64-haswell-optimised";
value = scopeCrossStatic.main.override {
x86_64_haswell_target_optimised =
if (crossSystem == "x86_64-linux-gnu" || crossSystem == "x86_64-linux-musl") then true else false;
};
}
# An output for a statically-linked unstripped debug ("dev") binary
{
name = "${binaryName}-debug";
value = scopeCrossStatic.main.override {
profile = "dev";
# debug build users expect full logs
disable_release_max_log_level = true;
};
}
# An output for a statically-linked unstripped debug binary with the
# "test" profile (for CI usage only)
{
name = "${binaryName}-test";
value = scopeCrossStatic.main.override {
profile = "test";
disable_release_max_log_level = true;
disable_features = [
# dont include experimental features
"experimental"
# this is non-functional on nix for some reason
"hardened_malloc"
# conduwuit_mods is a development-only hot reload feature
"conduwuit_mods"
];
};
}
# An output for a statically-linked binary with `--all-features`
{
name = "${binaryName}-all-features";
value = scopeCrossStatic.main.override {
all_features = true;
disable_features = [
# dont include experimental features
"experimental"
# jemalloc profiling/stats features are expensive and shouldn't
# be expected on non-debug builds.
"jemalloc_prof"
"jemalloc_stats"
# this is non-functional on nix for some reason
"hardened_malloc"
# conduwuit_mods is a development-only hot reload feature
"conduwuit_mods"
];
};
}
# An output for a statically-linked binary with `--all-features` and with x86_64 haswell
# target optimisations
{
name = "${binaryName}-all-features-x86_64-haswell-optimised";
value = scopeCrossStatic.main.override {
all_features = true;
disable_features = [
# dont include experimental features
"experimental"
# jemalloc profiling/stats features are expensive and shouldn't
# be expected on non-debug builds.
"jemalloc_prof"
"jemalloc_stats"
# this is non-functional on nix for some reason
"hardened_malloc"
# conduwuit_mods is a development-only hot reload feature
"conduwuit_mods"
];
x86_64_haswell_target_optimised =
if (crossSystem == "x86_64-linux-gnu" || crossSystem == "x86_64-linux-musl") then true else false;
};
}
# An output for a statically-linked unstripped debug ("dev") binary with `--all-features`
{
name = "${binaryName}-all-features-debug";
value = scopeCrossStatic.main.override {
profile = "dev";
all_features = true;
# debug build users expect full logs
disable_release_max_log_level = true;
disable_features = [
# dont include experimental features
"experimental"
# this is non-functional on nix for some reason
"hardened_malloc"
# conduwuit_mods is a development-only hot reload feature
"conduwuit_mods"
];
};
}
# An output for a statically-linked binary with hardened_malloc
{
name = "${binaryName}-hmalloc";
value = scopeCrossStatic.main.override {
features = [ "hardened_malloc" ];
};
}
]
)
[
#"x86_64-apple-darwin"
#"aarch64-apple-darwin"
"x86_64-linux-gnu"
"x86_64-linux-musl"
"aarch64-linux-musl"
]
)
);
}
);
inputs@{ flake-parts, ... }:
flake-parts.lib.mkFlake { inherit inputs; } {
imports = [ ./nix ];
systems = [
# good support
"x86_64-linux"
# support untested but theoretically there
"aarch64-linux"
];
};
}
+108
View File
@@ -0,0 +1,108 @@
{ inputs, ... }:
{
perSystem =
{
self',
lib,
pkgs,
...
}:
let
uwulib = inputs.self.uwulib.init pkgs;
rocksdbAllFeatures = self'.packages.rocksdb.override {
enableJemalloc = true;
enableLiburing = true;
};
commonAttrs = (uwulib.build.commonAttrs { }) // {
buildInputs = [
pkgs.liburing
pkgs.rust-jemalloc-sys-unprefixed
rocksdbAllFeatures
];
nativeBuildInputs = [
pkgs.pkg-config
# bindgen needs the build platform's libclang. Apparently due to "splicing
# weirdness", pkgs.rustPlatform.bindgenHook on its own doesn't quite do the
# right thing here.
pkgs.rustPlatform.bindgenHook
];
env = {
LIBCLANG_PATH = lib.makeLibraryPath [ pkgs.llvmPackages.libclang.lib ];
LD_LIBRARY_PATH = lib.makeLibraryPath [
pkgs.liburing
pkgs.rust-jemalloc-sys-unprefixed
rocksdbAllFeatures
];
}
// uwulib.environment.buildPackageEnv
// {
ROCKSDB_INCLUDE_DIR = "${rocksdbAllFeatures}/include";
ROCKSDB_LIB_DIR = "${rocksdbAllFeatures}/lib";
};
};
cargoArtifacts = self'.packages.continuwuity-all-features-deps;
in
{
# taken from
#
# https://crane.dev/examples/quick-start.html
checks = {
continuwuity-all-features-build = self'.packages.continuwuity-all-features-bin;
continuwuity-all-features-clippy = uwulib.build.craneLibForChecks.cargoClippy (
commonAttrs
// {
inherit cargoArtifacts;
cargoClippyExtraArgs = "-- --deny warnings";
}
);
continuwuity-all-features-docs = uwulib.build.craneLibForChecks.cargoDoc (
commonAttrs
// {
inherit cargoArtifacts;
# This can be commented out or tweaked as necessary, e.g. set to
# `--deny rustdoc::broken-intra-doc-links` to only enforce that lint
env.RUSTDOCFLAGS = "--deny warnings";
}
);
# Check formatting
continuwuity-all-features-fmt = uwulib.build.craneLibForChecks.cargoFmt {
src = uwulib.build.src;
};
continuwuity-all-features-toml-fmt = uwulib.build.craneLibForChecks.taploFmt {
src = pkgs.lib.sources.sourceFilesBySuffices uwulib.build.src [ ".toml" ];
# taplo arguments can be further customized below as needed
taploExtraArgs = "--config ${inputs.self}/taplo.toml";
};
# Audit dependencies
continuwuity-all-features-audit = uwulib.build.craneLibForChecks.cargoAudit {
inherit (inputs) advisory-db;
src = uwulib.build.src;
};
# Audit licenses
continuwuity-all-features-deny = uwulib.build.craneLibForChecks.cargoDeny {
src = uwulib.build.src;
};
# Run tests with cargo-nextest
# Consider setting `doCheck = false` on `continuwuity-all-features` if you do not want
# the tests to run twice
continuwuity-all-features-nextest = uwulib.build.craneLibForChecks.cargoNextest (
commonAttrs
// {
inherit cargoArtifacts;
partitions = 1;
partitionType = "count";
cargoNextestPartitionsExtraArgs = "--no-tests=pass";
}
);
};
};
}
+11
View File
@@ -0,0 +1,11 @@
{
imports = [
./checks
./packages
./shells
./tests
./hydra.nix
./fmt.nix
];
}
+37
View File
@@ -0,0 +1,37 @@
{ inputs, ... }:
{
# load the flake module from upstream
imports = [ inputs.treefmt-nix.flakeModule ];
perSystem =
{ self', lib, ... }:
{
treefmt = {
# repo root as project root
projectRoot = inputs.self;
# the formatters
programs = {
nixfmt.enable = true;
typos = {
enable = true;
configFile = "${inputs.self}/.typos.toml";
};
taplo = {
enable = true;
settings = lib.importTOML "${inputs.self}/taplo.toml";
};
};
settings.formatter.rustfmt = {
command = "${lib.getExe' self'.packages.dev-toolchain "rustfmt"}";
includes = [ "**/*.rs" ];
options = [
"--unstable-features"
"--edition=2024"
"--config-path=${inputs.self}/rustfmt.toml"
];
};
};
};
}
+9
View File
@@ -0,0 +1,9 @@
{ inputs, ... }:
let
lib = inputs.nixpkgs.lib;
in
{
flake.hydraJobs.packages = builtins.mapAttrs (
_name: lib.hydraJob
) inputs.self.packages.x86_64-linux;
}
+60
View File
@@ -0,0 +1,60 @@
{ inputs, ... }:
{
perSystem =
{
self',
lib,
pkgs,
...
}:
let
uwulib = inputs.self.uwulib.init pkgs;
in
{
packages =
lib.pipe
[
# this is the default variant
{
variantName = "default";
commonAttrsArgs.profile = "release";
rocksdb = self'.packages.rocksdb;
features = { };
}
# this is the variant with all features enabled (liburing + jemalloc)
{
variantName = "all-features";
commonAttrsArgs.profile = "release";
rocksdb = self'.packages.rocksdb.override {
enableJemalloc = true;
enableLiburing = true;
};
features = {
enabledFeatures = "all";
disabledFeatures = uwulib.features.defaultDisabledFeatures ++ [ "bindgen-static" ];
};
}
]
[
(builtins.map (cfg: rec {
deps = {
name = "continuwuity-${cfg.variantName}-deps";
value = uwulib.build.buildDeps {
features = uwulib.features.calcFeatures cfg.features;
inherit (cfg) commonAttrsArgs rocksdb;
};
};
bin = {
name = "continuwuity-${cfg.variantName}-bin";
value = uwulib.build.buildPackage {
deps = self'.packages.${deps.name};
features = uwulib.features.calcFeatures cfg.features;
inherit (cfg) commonAttrsArgs rocksdb;
};
};
}))
(builtins.concatMap builtins.attrValues)
builtins.listToAttrs
];
};
}
+14
View File
@@ -0,0 +1,14 @@
{
imports = [
./continuwuity
./rocksdb
./rust.nix
./uwulib
];
perSystem =
{ self', ... }:
{
packages.default = self'.packages.continuwuity-default-bin;
};
}
+12
View File
@@ -0,0 +1,12 @@
{
perSystem =
{
pkgs,
...
}:
{
packages = {
rocksdb = pkgs.callPackage ./package.nix { };
};
};
}
+88
View File
@@ -0,0 +1,88 @@
{
lib,
stdenv,
rocksdb,
liburing,
rust-jemalloc-sys-unprefixed,
enableJemalloc ? false,
enableLiburing ? false,
fetchFromGitea,
...
}:
let
notDarwin = !stdenv.hostPlatform.isDarwin;
in
(rocksdb.override {
# Override the liburing input for the build with our own so
# we have it built with the library flag
inherit liburing;
jemalloc = rust-jemalloc-sys-unprefixed;
# rocksdb fails to build with prefixed jemalloc, which is required on
# darwin due to [1]. In this case, fall back to building rocksdb with
# libc malloc. This should not cause conflicts, because all of the
# jemalloc symbols are prefixed.
#
# [1]: https://github.com/tikv/jemallocator/blob/ab0676d77e81268cd09b059260c75b38dbef2d51/jemalloc-sys/src/env.rs#L17
enableJemalloc = enableJemalloc && notDarwin;
# for some reason enableLiburing in nixpkgs rocksdb is default true
# which breaks Darwin entirely
enableLiburing = enableLiburing && notDarwin;
}).overrideAttrs
(old: {
src = fetchFromGitea {
domain = "forgejo.ellis.link";
owner = "continuwuation";
repo = "rocksdb";
rev = "10.5.fb";
sha256 = "sha256-X4ApGLkHF9ceBtBg77dimEpu720I79ffLoyPa8JMHaU=";
};
version = "10.5.fb";
cmakeFlags =
lib.subtractLists (builtins.map (flag: lib.cmakeBool flag true) [
# No real reason to have snappy or zlib, no one uses this
"WITH_SNAPPY"
"ZLIB"
"WITH_ZLIB"
# We don't need to use ldb or sst_dump (core_tools)
"WITH_CORE_TOOLS"
# We don't need to build rocksdb tests
"WITH_TESTS"
# We use rust-rocksdb via C interface and don't need C++ RTTI
"USE_RTTI"
# This doesn't exist in RocksDB, and USE_SSE is deprecated for
# PORTABLE=$(march)
"FORCE_SSE42"
]) old.cmakeFlags
++ (builtins.map (flag: lib.cmakeBool flag false) [
# No real reason to have snappy, no one uses this
"WITH_SNAPPY"
"ZLIB"
"WITH_ZLIB"
# We don't need to use ldb or sst_dump (core_tools)
"WITH_CORE_TOOLS"
# We don't need trace tools
"WITH_TRACE_TOOLS"
# We don't need to build rocksdb tests
"WITH_TESTS"
# We use rust-rocksdb via C interface and don't need C++ RTTI
"USE_RTTI"
]);
enableLiburing = enableLiburing && notDarwin;
# outputs has "tools" which we don't need or use
outputs = [ "out" ];
# preInstall hooks has stuff for messing with ldb/sst_dump which we don't need or use
preInstall = "";
# We have this already at https://forgejo.ellis.link/continuwuation/rocksdb/commit/a935c0273e1ba44eacf88ce3685a9b9831486155
# Unsetting `patches` so we don't have to revert it and make this nix exclusive
patches = [ ];
})
+32
View File
@@ -0,0 +1,32 @@
{ inputs, ... }:
{
perSystem =
{
system,
lib,
...
}:
{
packages =
let
fnx = inputs.fenix.packages.${system};
stable = fnx.fromToolchainFile {
file = inputs.self + "/rust-toolchain.toml";
# See also `rust-toolchain.toml`
sha256 = "sha256-SJwZ8g0zF2WrKDVmHrVG3pD2RGoQeo24MEXnNx5FyuI=";
};
in
{
# used for building nix stuff (doesn't include rustfmt overhead)
build-toolchain = stable;
# used for dev shells
dev-toolchain = fnx.combine [
stable
# use the nightly rustfmt because we use nightly features
fnx.complete.rustfmt
];
};
};
}
+108
View File
@@ -0,0 +1,108 @@
args@{ pkgs, inputs, ... }:
let
inherit (pkgs) lib;
uwuenv = import ./environment.nix args;
selfpkgs = inputs.self.packages.${pkgs.stdenv.system};
in
rec {
# basic, very minimal instance of the crane library with a minimal rust toolchain
craneLib = (inputs.crane.mkLib pkgs).overrideToolchain (_: selfpkgs.build-toolchain);
# the checks require more rust toolchain components, hence we have this separate instance of the crane library
craneLibForChecks = (inputs.crane.mkLib pkgs).overrideToolchain (_: selfpkgs.dev-toolchain);
# meta information (name, version, etc) of the rust crate based on the Cargo.toml
crateInfo = craneLib.crateNameFromCargoToml { cargoToml = "${inputs.self}/Cargo.toml"; };
src =
let
# see https://crane.dev/API.html#cranelibfiltercargosources
#
# we need to keep the `web` directory which would be filtered out by the regular source filtering function
#
# https://crane.dev/API.html#cranelibcleancargosource
isWebTemplate = path: _type: builtins.match ".*src/web.*" path != null;
isRust = craneLib.filterCargoSources;
isNix = path: _type: builtins.match ".+/nix.*" path != null;
webOrRustNotNix = p: t: !(isNix p t) && (isWebTemplate p t || isRust p t);
in
lib.cleanSourceWith {
src = inputs.self;
filter = webOrRustNotNix;
name = "source";
};
# common attrs that are shared between building continuwuity's deps and the package itself
commonAttrs =
{
profile ? "dev",
...
}:
{
inherit (crateInfo)
pname
version
;
inherit src;
# this prevents unnecessary rebuilds
strictDeps = true;
dontStrip = profile == "dev" || profile == "test";
dontPatchELF = profile == "dev" || profile == "test";
doCheck = true;
nativeBuildInputs = [
# bindgen needs the build platform's libclang. Apparently due to "splicing
# weirdness", pkgs.rustPlatform.bindgenHook on its own doesn't quite do the
# right thing here.
pkgs.rustPlatform.bindgenHook
];
};
makeRocksDBEnv =
{ rocksdb }:
{
ROCKSDB_INCLUDE_DIR = "${rocksdb}/include";
ROCKSDB_LIB_DIR = "${rocksdb}/lib";
};
# function that builds the continuwuity dependencies derivation
buildDeps =
{
rocksdb,
features,
commonAttrsArgs,
}:
craneLib.buildDepsOnly (
(commonAttrs commonAttrsArgs)
// {
env = uwuenv.buildDepsOnlyEnv // (makeRocksDBEnv { inherit rocksdb; });
inherit (features) cargoExtraArgs;
}
);
# function that builds the continuwuity package
buildPackage =
{
deps,
rocksdb,
features,
commonAttrsArgs,
}:
let
rocksdbEnv = makeRocksDBEnv { inherit rocksdb; };
in
craneLib.buildPackage (
(commonAttrs commonAttrsArgs)
// {
cargoArtifacts = deps;
doCheck = true;
env = uwuenv.buildPackageEnv // rocksdbEnv;
passthru.env = uwuenv.buildPackageEnv // rocksdbEnv;
meta.mainProgram = crateInfo.pname;
inherit (features) cargoExtraArgs;
}
);
}
+10
View File
@@ -0,0 +1,10 @@
{ inputs, ... }:
{
flake.uwulib = {
init = pkgs: {
features = import ./features.nix { inherit pkgs inputs; };
environment = import ./environment.nix { inherit pkgs inputs; };
build = import ./build.nix { inherit pkgs inputs; };
};
};
}
+18
View File
@@ -0,0 +1,18 @@
args@{ pkgs, inputs, ... }:
let
uwubuild = import ./build.nix args;
in
rec {
buildDepsOnlyEnv = {
# https://crane.dev/faq/rebuilds-bindgen.html
NIX_OUTPATH_USED_AS_RANDOM_SEED = "aaaaaaaaaa";
CARGO_PROFILE = "release";
}
// uwubuild.craneLib.mkCrossToolchainEnv (p: pkgs.clangStdenv);
buildPackageEnv = {
GIT_COMMIT_HASH = inputs.self.rev or inputs.self.dirtyRev or "";
GIT_COMMIT_HASH_SHORT = inputs.self.shortRev or inputs.self.dirtyShortRev or "";
}
// buildDepsOnlyEnv;
}
+77
View File
@@ -0,0 +1,77 @@
{ pkgs, inputs, ... }:
let
inherit (pkgs) lib;
in
rec {
defaultDisabledFeatures = [
# dont include experimental features
"experimental"
# jemalloc profiling/stats features are expensive and shouldn't
# be expected on non-debug builds.
"jemalloc_prof"
"jemalloc_stats"
# this is non-functional on nix for some reason
"hardened_malloc"
# conduwuit_mods is a development-only hot reload feature
"conduwuit_mods"
# we don't want to enable this feature set by default but be more specific about it
"full"
];
# We perform default-feature unification in nix, because some of the dependencies
# on the nix side depend on feature values.
calcFeatures =
{
tomlPath ? "${inputs.self}/src/main",
# either a list of feature names or a string "all" which enables all non-default features
enabledFeatures ? [ ],
disabledFeatures ? defaultDisabledFeatures,
default_features ? true,
disable_release_max_log_level ? false,
}:
let
# simple helper to get the contents of a Cargo.toml file in a nix format
getToml = path: lib.importTOML "${path}/Cargo.toml";
# get all the features except for the default features
allFeatures = lib.pipe tomlPath [
getToml
(manifest: manifest.features)
lib.attrNames
(lib.remove "default")
];
# get just the default enabled features
allDefaultFeatures = lib.pipe tomlPath [
getToml
(manifest: manifest.features.default)
];
# depending on the value of enabledFeatures choose just a set or all non-default features
#
# - [ list of features ] -> choose exactly the features listed
# - "all" -> choose all non-default features
additionalFeatures = if enabledFeatures == "all" then allFeatures else enabledFeatures;
# unification with default features (if enabled)
features = lib.unique (additionalFeatures ++ lib.optionals default_features allDefaultFeatures);
# prepare the features that are subtracted from the set
disabledFeatures' =
disabledFeatures ++ lib.optionals disable_release_max_log_level [ "release_max_log_level" ];
# construct the final feature set
finalFeatures = lib.subtractLists disabledFeatures' features;
in
{
# final feature set, useful for querying it
features = finalFeatures;
# crane flag with the relevant features
cargoExtraArgs = builtins.concatStringsSep " " [
"--no-default-features"
"--locked"
(lib.optionalString (finalFeatures != [ ]) "--features")
(builtins.concatStringsSep "," finalFeatures)
];
};
}
+29
View File
@@ -0,0 +1,29 @@
{ inputs, ... }:
{
perSystem =
{
self',
lib,
pkgs,
...
}:
let
uwulib = inputs.self.uwulib.init pkgs;
rocksdbAllFeatures = self'.packages.rocksdb.override {
enableJemalloc = true;
enableLiburing = true;
};
in
{
# basic nix shell containing all things necessary to build continuwuity in all flavors manually (on x86_64-linux)
devShells.default = uwulib.build.craneLib.devShell {
packages = [
pkgs.pkg-config
pkgs.liburing
pkgs.rust-jemalloc-sys-unprefixed
rocksdbAllFeatures
];
env.LIBCLANG_PATH = lib.makeLibraryPath [ pkgs.llvmPackages.libclang.lib ];
};
};
}
+124
View File
@@ -0,0 +1,124 @@
{
perSystem =
{
self',
lib,
pkgs,
...
}:
{
# run some nixos tests as checks
checks = lib.pipe self'.packages [
# we take all packages (names)
builtins.attrNames
# we filter out all packages that end with `-bin` (which we are interested in for testing)
(builtins.filter (lib.hasSuffix "-bin"))
# for each of these binaries we built the basic nixos test
#
# this test was initially yoinked from
#
# https://github.com/NixOS/nixpkgs/blob/960ce26339661b1b69c6f12b9063ca51b688615f/nixos/tests/matrix/continuwuity.nix
(builtins.map (name: {
name = "test-${name}";
value = pkgs.testers.runNixOSTest {
inherit name;
nodes = {
continuwuity = {
services.matrix-continuwuity = {
enable = true;
package = self'.packages.${name};
settings.global = {
server_name = name;
address = [ "0.0.0.0" ];
allow_registration = true;
yes_i_am_very_very_sure_i_want_an_open_registration_server_prone_to_abuse = true;
};
extraEnvironment.RUST_BACKTRACE = "yes";
};
networking.firewall.allowedTCPPorts = [ 6167 ];
};
client =
{ pkgs, ... }:
{
environment.systemPackages = [
(pkgs.writers.writePython3Bin "do_test" { libraries = [ pkgs.python3Packages.matrix-nio ]; } ''
import asyncio
import nio
async def main() -> None:
# Connect to continuwuity
client = nio.AsyncClient("http://continuwuity:6167", "alice")
# Register as user alice
response = await client.register("alice", "my-secret-password")
# Log in as user alice
response = await client.login("my-secret-password")
# Create a new room
response = await client.room_create(federate=False)
print("Matrix room create response:", response)
assert isinstance(response, nio.RoomCreateResponse)
room_id = response.room_id
# Join the room
response = await client.join(room_id)
print("Matrix join response:", response)
assert isinstance(response, nio.JoinResponse)
# Send a message to the room
response = await client.room_send(
room_id=room_id,
message_type="m.room.message",
content={
"msgtype": "m.text",
"body": "Hello continuwuity!"
}
)
print("Matrix room send response:", response)
assert isinstance(response, nio.RoomSendResponse)
# Sync responses
response = await client.sync(timeout=30000)
print("Matrix sync response:", response)
assert isinstance(response, nio.SyncResponse)
# Check the message was received by continuwuity
last_message = response.rooms.join[room_id].timeline.events[-1].body
assert last_message == "Hello continuwuity!"
# Leave the room
response = await client.room_leave(room_id)
print("Matrix room leave response:", response)
assert isinstance(response, nio.RoomLeaveResponse)
# Close the client
await client.close()
if __name__ == "__main__":
asyncio.run(main())
'')
];
};
};
testScript = ''
start_all()
with subtest("start continuwuity"):
continuwuity.wait_for_unit("continuwuity.service")
continuwuity.wait_for_open_port(6167)
with subtest("ensure messages can be exchanged"):
client.succeed("do_test >&2")
'';
};
}))
builtins.listToAttrs
];
};
}
+8 -6
View File
@@ -12,13 +12,14 @@ Group=conduwuit
Type=notify-reload
ReloadSignal=SIGUSR1
Environment="CONTINUWUITY_CONFIG=/etc/conduwuit/conduwuit.toml"
Environment="CONTINUWUITY_LOG_TO_JOURNALD=true"
Environment="CONTINUWUITY_JOURNALD_IDENTIFIER=%N"
Environment="CONTINUWUITY_DATABASE_PATH=/var/lib/conduwuit"
Environment="CONTINUWUITY_DATABASE_PATH=%S/conduwuit"
Environment="CONTINUWUITY_CONFIG_RELOAD_SIGNAL=true"
ExecStart=/usr/bin/conduwuit
LoadCredential=conduwuit.toml:/etc/conduwuit/conduwuit.toml
ExecStart=/usr/bin/conduwuit --config ${CREDENTIALS_DIRECTORY}/conduwuit.toml
AmbientCapabilities=
CapabilityBoundingSet=
@@ -52,8 +53,9 @@ SystemCallFilter=@system-service @resources
SystemCallFilter=~@clock @debug @module @mount @reboot @swap @cpu-emulation @obsolete @timer @chown @setuid @privileged @keyring @ipc
SystemCallErrorNumber=EPERM
# ConfigurationDirectory isn't specified here because it's created by
# the distro's package manager.
StateDirectory=conduwuit
ConfigurationDirectory=conduwuit
RuntimeDirectory=conduwuit
RuntimeDirectoryMode=0750
@@ -61,7 +63,7 @@ Restart=on-failure
RestartSec=5
TimeoutStopSec=4m
TimeoutStartSec=4m
TimeoutStartSec=10m
StartLimitInterval=1m
StartLimitBurst=5
+2 -2
View File
@@ -51,7 +51,7 @@ find .cargo/registry/ -executable -name "*.rs" -exec chmod -x {} +
%install
install -Dpm0755 target/rpm/conduwuit -t %{buildroot}%{_bindir}
install -Dpm0644 pkg/conduwuit.service -t %{buildroot}%{_unitdir}
install -Dpm0644 conduwuit-example.toml %{buildroot}%{_sysconfdir}/conduwuit/conduwuit.toml
install -Dpm0600 conduwuit-example.toml %{buildroot}%{_sysconfdir}/conduwuit/conduwuit.toml
%files
%license LICENSE
@@ -60,7 +60,7 @@ install -Dpm0644 conduwuit-example.toml %{buildroot}%{_sysconfdir}/conduwuit/con
%doc CONTRIBUTING.md
%doc README.md
%doc SECURITY.md
%config %{_sysconfdir}/conduwuit/conduwuit.toml
%config(noreplace) %{_sysconfdir}/conduwuit/conduwuit.toml
%{_bindir}/conduwuit
%{_unitdir}/conduwuit.service
@@ -1,83 +0,0 @@
{ lib
, pkgsBuildHost
, rust
, stdenv
}:
lib.optionalAttrs stdenv.hostPlatform.isStatic
{
ROCKSDB_STATIC = "";
}
//
{
CARGO_BUILD_RUSTFLAGS =
lib.concatStringsSep
" "
(lib.optionals
stdenv.hostPlatform.isStatic
[ "-C" "relocation-model=static" ]
++ lib.optionals
(stdenv.buildPlatform.config != stdenv.hostPlatform.config)
[
"-l"
"c"
"-l"
"stdc++"
"-L"
"${stdenv.cc.cc.lib}/${stdenv.hostPlatform.config}/lib"
]
);
}
# What follows is stolen from [here][0]. Its purpose is to properly
# configure compilers and linkers for various stages of the build, and
# even covers the case of build scripts that need native code compiled and
# run on the build platform (I think).
#
# [0]: https://github.com/NixOS/nixpkgs/blob/nixpkgs-unstable/pkgs/build-support/rust/lib/default.nix#L48-L68
//
(
let
inherit (rust.lib) envVars;
in
lib.optionalAttrs
(stdenv.targetPlatform.rust.rustcTarget
!= stdenv.hostPlatform.rust.rustcTarget)
(
let
inherit (stdenv.targetPlatform.rust) cargoEnvVarTarget;
in
{
"CC_${cargoEnvVarTarget}" = envVars.ccForTarget;
"CXX_${cargoEnvVarTarget}" = envVars.cxxForTarget;
"CARGO_TARGET_${cargoEnvVarTarget}_LINKER" = envVars.ccForTarget;
}
)
//
(
let
inherit (stdenv.hostPlatform.rust) cargoEnvVarTarget rustcTarget;
in
{
"CC_${cargoEnvVarTarget}" = envVars.ccForHost;
"CXX_${cargoEnvVarTarget}" = envVars.cxxForHost;
"CARGO_TARGET_${cargoEnvVarTarget}_LINKER" = envVars.ccForHost;
CARGO_BUILD_TARGET = rustcTarget;
}
)
//
(
let
inherit (stdenv.buildPlatform.rust) cargoEnvVarTarget;
in
{
"CC_${cargoEnvVarTarget}" = envVars.ccForBuild;
"CXX_${cargoEnvVarTarget}" = envVars.cxxForBuild;
"CARGO_TARGET_${cargoEnvVarTarget}_LINKER" = envVars.ccForBuild;
HOST_CC = "${pkgsBuildHost.stdenv.cc}/bin/cc";
HOST_CXX = "${pkgsBuildHost.stdenv.cc}/bin/c++";
}
)
)
-224
View File
@@ -1,224 +0,0 @@
# Dependencies (keep sorted)
{ craneLib
, inputs
, jq
, lib
, libiconv
, liburing
, pkgsBuildHost
, rocksdb
, removeReferencesTo
, rust
, rust-jemalloc-sys
, stdenv
# Options (keep sorted)
, all_features ? false
, default_features ? true
# default list of disabled features
, disable_features ? [
# dont include experimental features
"experimental"
# jemalloc profiling/stats features are expensive and shouldn't
# be expected on non-debug builds.
"jemalloc_prof"
"jemalloc_stats"
# this is non-functional on nix for some reason
"hardened_malloc"
# conduwuit_mods is a development-only hot reload feature
"conduwuit_mods"
]
, disable_release_max_log_level ? false
, features ? [ ]
, profile ? "release"
# rocksdb compiled with -march=haswell and target-cpu=haswell rustflag
# haswell is pretty much any x86 cpu made in the last 12 years, and
# supports modern CPU extensions that rocksdb can make use of.
# disable if trying to make a portable x86_64 build for very old hardware
, x86_64_haswell_target_optimised ? false
}:
let
# We perform default-feature unification in nix, because some of the dependencies
# on the nix side depend on feature values.
crateFeatures = path:
let manifest = lib.importTOML "${path}/Cargo.toml"; in
lib.remove "default" (lib.attrNames manifest.features);
crateDefaultFeatures = path:
(lib.importTOML "${path}/Cargo.toml").features.default;
allDefaultFeatures = crateDefaultFeatures "${inputs.self}/src/main";
allFeatures = crateFeatures "${inputs.self}/src/main";
features' = lib.unique
(features ++
lib.optionals default_features allDefaultFeatures ++
lib.optionals all_features allFeatures);
disable_features' = disable_features ++ lib.optionals disable_release_max_log_level [ "release_max_log_level" ];
features'' = lib.subtractLists disable_features' features';
featureEnabled = feature: builtins.elem feature features'';
enableLiburing = featureEnabled "io_uring" && !stdenv.hostPlatform.isDarwin;
# This derivation will set the JEMALLOC_OVERRIDE variable, causing the
# tikv-jemalloc-sys crate to use the nixpkgs jemalloc instead of building it's
# own. In order for this to work, we need to set flags on the build that match
# whatever flags tikv-jemalloc-sys was going to use. These are dependent on
# which features we enable in tikv-jemalloc-sys.
rust-jemalloc-sys' = (rust-jemalloc-sys.override {
# tikv-jemalloc-sys/unprefixed_malloc_on_supported_platforms feature
unprefixed = true;
}).overrideAttrs (old: {
configureFlags = old.configureFlags ++
# we dont need docs
[ "--disable-doc" ] ++
# we dont need cxx/C++ integration
[ "--disable-cxx" ] ++
# tikv-jemalloc-sys/profiling feature
lib.optional (featureEnabled "jemalloc_prof") "--enable-prof" ++
# tikv-jemalloc-sys/stats feature
(if (featureEnabled "jemalloc_stats") then [ "--enable-stats" ] else [ "--disable-stats" ]);
});
buildDepsOnlyEnv =
let
rocksdb' = (rocksdb.override {
jemalloc = lib.optional (featureEnabled "jemalloc") rust-jemalloc-sys';
# rocksdb fails to build with prefixed jemalloc, which is required on
# darwin due to [1]. In this case, fall back to building rocksdb with
# libc malloc. This should not cause conflicts, because all of the
# jemalloc symbols are prefixed.
#
# [1]: https://github.com/tikv/jemallocator/blob/ab0676d77e81268cd09b059260c75b38dbef2d51/jemalloc-sys/src/env.rs#L17
enableJemalloc = featureEnabled "jemalloc" && !stdenv.hostPlatform.isDarwin;
# for some reason enableLiburing in nixpkgs rocksdb is default true
# which breaks Darwin entirely
inherit enableLiburing;
}).overrideAttrs (old: {
inherit enableLiburing;
cmakeFlags = (if x86_64_haswell_target_optimised then
(lib.subtractLists [
# dont make a portable build if x86_64_haswell_target_optimised is enabled
"-DPORTABLE=1"
]
old.cmakeFlags
++ [ "-DPORTABLE=haswell" ]) else [ "-DPORTABLE=1" ]
)
++ old.cmakeFlags;
# outputs has "tools" which we dont need or use
outputs = [ "out" ];
# preInstall hooks has stuff for messing with ldb/sst_dump which we dont need or use
preInstall = "";
});
in
{
# https://crane.dev/faq/rebuilds-bindgen.html
NIX_OUTPATH_USED_AS_RANDOM_SEED = "aaaaaaaaaa";
CARGO_PROFILE = profile;
ROCKSDB_INCLUDE_DIR = "${rocksdb'}/include";
ROCKSDB_LIB_DIR = "${rocksdb'}/lib";
}
//
(import ./cross-compilation-env.nix {
# Keep sorted
inherit
lib
pkgsBuildHost
rust
stdenv;
});
buildPackageEnv = {
GIT_COMMIT_HASH = inputs.self.rev or inputs.self.dirtyRev or "";
GIT_COMMIT_HASH_SHORT = inputs.self.shortRev or inputs.self.dirtyShortRev or "";
} // buildDepsOnlyEnv // {
# Only needed in static stdenv because these are transitive dependencies of rocksdb
CARGO_BUILD_RUSTFLAGS = buildDepsOnlyEnv.CARGO_BUILD_RUSTFLAGS
+ lib.optionalString (enableLiburing && stdenv.hostPlatform.isStatic)
" -L${lib.getLib liburing}/lib -luring"
+ lib.optionalString x86_64_haswell_target_optimised
" -Ctarget-cpu=haswell";
};
commonAttrs = {
inherit
(craneLib.crateNameFromCargoToml {
cargoToml = "${inputs.self}/Cargo.toml";
})
pname
version;
src = let filter = inputs.nix-filter.lib; in filter {
root = inputs.self;
# Keep sorted
include = [
".cargo"
"Cargo.lock"
"Cargo.toml"
"src"
"xtask"
];
};
doCheck = true;
cargoExtraArgs = "--no-default-features --locked "
+ lib.optionalString
(features'' != [ ])
"--features " + (builtins.concatStringsSep "," features'');
dontStrip = profile == "dev" || profile == "test";
dontPatchELF = profile == "dev" || profile == "test";
buildInputs = lib.optional (featureEnabled "jemalloc") rust-jemalloc-sys'
# needed to build Rust applications on macOS
++ lib.optionals stdenv.hostPlatform.isDarwin [
# https://github.com/NixOS/nixpkgs/issues/206242
# ld: library not found for -liconv
libiconv
# https://stackoverflow.com/questions/69869574/properly-adding-darwin-apple-sdk-to-a-nix-shell
# https://discourse.nixos.org/t/compile-a-rust-binary-on-macos-dbcrossbar/8612
pkgsBuildHost.darwin.apple_sdk.frameworks.Security
];
nativeBuildInputs = [
# bindgen needs the build platform's libclang. Apparently due to "splicing
# weirdness", pkgs.rustPlatform.bindgenHook on its own doesn't quite do the
# right thing here.
pkgsBuildHost.rustPlatform.bindgenHook
# We don't actually depend on `jq`, but crane's `buildPackage` does, but
# its `buildDepsOnly` doesn't. This causes those two derivations to have
# differing values for `NIX_CFLAGS_COMPILE`, which contributes to spurious
# rebuilds of bindgen and its depedents.
jq
];
};
in
craneLib.buildPackage (commonAttrs // {
cargoArtifacts = craneLib.buildDepsOnly (commonAttrs // {
env = buildDepsOnlyEnv;
});
doCheck = true;
cargoExtraArgs = "--no-default-features --locked "
+ lib.optionalString
(features'' != [ ])
"--features " + (builtins.concatStringsSep "," features'');
env = buildPackageEnv;
passthru = {
env = buildPackageEnv;
};
meta.mainProgram = commonAttrs.pname;
})
+11 -11
View File
@@ -10,15 +10,19 @@
"nix": {
"enabled": true
},
"pre-commit": {
"enabled": true
},
"labels": ["Dependencies", "Dependencies/Renovate"],
"ignoreDeps": [
"tikv-jemallocator",
"tikv-jemalloc-sys",
"tikv-jemalloc-ctl",
"opentelemetry",
"opentelemetry_sdk",
"opentelemetry-jaeger",
"tracing-opentelemetry"
"rustyline-async",
"event-listener",
"async-channel",
"core_affinity",
"hyper-util"
],
"github-actions": {
"enabled": true,
@@ -64,12 +68,8 @@
"matchDatasources": ["docker"],
"matchPackageNames": ["ghcr.io/renovatebot/renovate"],
"automerge": true,
"automergeStrategy": "fast-forward"
},
{
"description": "Group lockfile updates into a single PR",
"matchUpdateTypes": ["lockFileMaintenance"],
"groupName": "lockfile-maintenance"
"automergeStrategy": "fast-forward",
"extends": ["schedule:earlyMondays"]
}
],
"customManagers": [
@@ -81,7 +81,7 @@
"/(^|/|\\.)([Dd]ocker|[Cc]ontainer)file$/"
],
"matchStrings": [
"# renovate: datasource=(?<datasource>[a-z-.]+?) depName=(?<depName>[^\\s]+?)(?: (lookupName|packageName)=(?<packageName>[^\\s]+?))?(?: versioning=(?<versioning>[^\\s]+?))?(?: extractVersion=(?<extractVersion>[^\\s]+?))?(?: registryUrl=(?<registryUrl>[^\\s]+?))?\\s+(?:ENV|ARG)\\s+[A-Za-z0-9_]+?_VERSION[ =][\"']?(?<currentValue>.+?)[\"']?\\s"
"# renovate: datasource=(?<datasource>[a-zA-Z0-9-._]+?) depName=(?<depName>[^\\s]+?)(?: (lookupName|packageName)=(?<packageName>[^\\s]+?))?(?: versioning=(?<versioning>[^\\s]+?))?(?: extractVersion=(?<extractVersion>[^\\s]+?))?(?: registryUrl=(?<registryUrl>[^\\s]+?))?\\s+(?:ENV\\s+|ARG\\s+)?[A-Za-z0-9_]+?_VERSION[ =][\"']?(?<currentValue>.+?)[\"']?\\s+(?:(?:ENV\\s+|ARG\\s+)?[A-Za-z0-9_]+?_CHECKSUM[ =][\"']?(?<currentDigest>.+?)[\"']?\\s)?"
]
}
]
+1 -1
View File
@@ -10,7 +10,7 @@
[toolchain]
profile = "minimal"
channel = "1.89.0"
channel = "1.90.0"
components = [
# For rust-analyzer
"rust-src",
+1 -1
View File
@@ -85,7 +85,7 @@ futures.workspace = true
log.workspace = true
ruma.workspace = true
serde_json.workspace = true
serde_yml.workspace = true
serde-saphyr.workspace = true
tokio.workspace = true
tracing-subscriber.workspace = true
tracing.workspace = true
+2 -2
View File
@@ -16,7 +16,7 @@ pub(super) async fn register(&self) -> Result {
let range = 1..checked!(body_len - 1)?;
let appservice_config_body = body[range].join("\n");
let parsed_config = serde_yml::from_str(&appservice_config_body);
let parsed_config = serde_saphyr::from_str(&appservice_config_body);
match parsed_config {
| Err(e) => return Err!("Could not parse appservice config as YAML: {e}"),
| Ok(registration) => match self
@@ -57,7 +57,7 @@ pub(super) async fn show_appservice_config(&self, appservice_identifier: String)
{
| None => return Err!("Appservice does not exist."),
| Some(config) => {
let config_str = serde_yml::to_string(&config)?;
let config_str = serde_saphyr::to_string(&config)?;
write!(self, "Config for {appservice_identifier}:\n\n```yaml\n{config_str}\n```")
},
}
+12 -6
View File
@@ -2,7 +2,8 @@
use conduwuit::{
Err, Result, debug, debug_info, debug_warn, error, info, trace,
utils::time::parse_timepoint_ago, warn,
utils::time::{TimeDirection, parse_timepoint_ago},
warn,
};
use conduwuit_service::media::Dim;
use ruma::{Mxc, OwnedEventId, OwnedMxcUri, OwnedServerName};
@@ -235,14 +236,19 @@ pub(super) async fn delete_past_remote_media(
}
assert!(!(before && after), "--before and --after should not be specified together");
let duration = parse_timepoint_ago(&duration)?;
let direction = if after {
TimeDirection::After
} else {
TimeDirection::Before
};
let time_boundary = parse_timepoint_ago(&duration)?;
let deleted_count = self
.services
.media
.delete_all_remote_media_at_after_time(
duration,
before,
after,
.delete_all_media_within_timeframe(
time_boundary,
direction,
yes_i_want_to_delete_local_media,
)
.await?;
+17 -5
View File
@@ -27,12 +27,24 @@ pub enum MediaCommand {
/// filesystem. This will always ignore errors.
DeleteList,
/// - Deletes all remote (and optionally local) media created before or
/// after [duration] time using filesystem metadata first created at date,
/// or fallback to last modified date. This will always ignore errors by
/// default.
/// Deletes all remote (and optionally local) media created before/after
/// [duration] ago, using filesystem metadata first created at date, or
/// fallback to last modified date. This will always ignore errors by
/// default.
///
/// * Examples:
/// * Delete all remote media older than a year:
///
/// `!admin media delete-past-remote-media -b 1y`
///
/// * Delete all remote and local media from 3 days ago, up until now:
///
/// `!admin media delete-past-remote-media -a 3d
/// --yes-i-want-to-delete-local-media`
#[command(verbatim_doc_comment)]
DeletePastRemoteMedia {
/// - The relative time (e.g. 30s, 5m, 7d) within which to search
/// - The relative time (e.g. 30s, 5m, 7d) from now within which to
/// search
duration: String,
/// - Only delete media created before [duration] ago
+1 -1
View File
@@ -41,7 +41,7 @@ async fn changes_since(
let results: Vec<_> = self
.services
.account_data
.changes_since(room_id.as_deref(), &user_id, since, None)
.changes_since(room_id.as_deref(), &user_id, Some(since), None)
.collect()
.await;
let query_time = timer.elapsed();
+2 -2
View File
@@ -389,7 +389,7 @@ pub(crate) async fn get_key_changes_route(
device_list_updates.extend(
services
.users
.keys_changed(sender_user, from, Some(to))
.keys_changed(sender_user, Some(from), Some(to))
.map(ToOwned::to_owned)
.collect::<Vec<_>>()
.await,
@@ -401,7 +401,7 @@ pub(crate) async fn get_key_changes_route(
device_list_updates.extend(
services
.users
.room_keys_changed(room_id, from, Some(to))
.room_keys_changed(room_id, Some(from), Some(to))
.map(|(user_id, _)| user_id)
.map(ToOwned::to_owned)
.collect::<Vec<_>>()
+6 -2
View File
@@ -64,10 +64,14 @@ pub(crate) async fn create_content_route(
media_id: &utils::random_string(MXC_LENGTH),
};
services
if let Err(e) = services
.media
.create(mxc, Some(user), Some(&content_disposition), content_type, &body.file)
.await?;
.await
{
err!("Failed to save uploaded media: {e}");
return Err!(Request(Unknown("Failed to save uploaded media")));
}
let blurhash = body.generate_blurhash.then(|| {
services
+9 -24
View File
@@ -5,7 +5,7 @@
use conduwuit::{
Err, Result, debug, debug_info, debug_warn, err, info,
matrix::{
event::{Event, gen_event_id},
event::gen_event_id,
pdu::{PduBuilder, PduEvent},
},
result::FlatOk,
@@ -458,7 +458,7 @@ async fn knock_room_helper_local(
.await,
};
let send_knock_response = services
services
.sending
.send_federation_request(&remote_server, send_knock_request)
.await?;
@@ -477,20 +477,14 @@ async fn knock_room_helper_local(
.map_err(|e| err!(BadServerResponse("Invalid knock event PDU: {e:?}")))?;
info!("Updating membership locally to knock state with provided stripped state events");
// TODO: this call does not appear to do anything because `update_membership`
// doesn't call `mark_as_knock`. investigate further, ideally with the aim of
// removing this call entirely -- Ginger thinks `update_membership` should only
// be called from `force_state` and `append_pdu`.
services
.rooms
.state_cache
.update_membership(
room_id,
sender_user,
parsed_knock_pdu
.get_content::<RoomMemberEventContent>()
.expect("we just created this"),
sender_user,
Some(send_knock_response.knock_room_state),
None,
false,
)
.update_membership(room_id, sender_user, &parsed_knock_pdu, false)
.await?;
info!("Appending room knock event locally");
@@ -677,20 +671,11 @@ async fn knock_room_helper_remote(
.await?;
info!("Updating membership locally to knock state with provided stripped state events");
// TODO: see TODO on the other call to `update_membership`
services
.rooms
.state_cache
.update_membership(
room_id,
sender_user,
parsed_knock_pdu
.get_content::<RoomMemberEventContent>()
.expect("we just created this"),
sender_user,
Some(send_knock_response.knock_room_state),
None,
false,
)
.update_membership(room_id, sender_user, &parsed_knock_pdu, false)
.await?;
info!("Appending room knock event locally");
+97 -106
View File
@@ -2,12 +2,12 @@
use axum::extract::State;
use conduwuit::{
Err, Result, debug_info, debug_warn, err,
Err, Pdu, Result, debug_info, debug_warn, err,
matrix::{event::gen_event_id, pdu::PduBuilder},
utils::{self, FutureBoolExt, future::ReadyEqExt},
warn,
};
use futures::{FutureExt, StreamExt, TryFutureExt, pin_mut};
use futures::{FutureExt, StreamExt, pin_mut};
use ruma::{
CanonicalJsonObject, CanonicalJsonValue, OwnedServerName, RoomId, RoomVersionId, UserId,
api::{
@@ -81,42 +81,9 @@ pub async fn leave_room(
room_id: &RoomId,
reason: Option<String>,
) -> Result {
let default_member_content = RoomMemberEventContent {
membership: MembershipState::Leave,
reason: reason.clone(),
join_authorized_via_users_server: None,
is_direct: None,
avatar_url: None,
displayname: None,
third_party_invite: None,
blurhash: None,
redact_events: None,
};
let is_banned = services.rooms.metadata.is_banned(room_id);
let is_disabled = services.rooms.metadata.is_disabled(room_id);
pin_mut!(is_banned, is_disabled);
if is_banned.or(is_disabled).await {
// the room is banned/disabled, the room must be rejected locally since we
// cant/dont want to federate with this server
services
.rooms
.state_cache
.update_membership(
room_id,
user_id,
default_member_content,
user_id,
None,
None,
true,
)
.await?;
return Ok(());
}
let dont_have_room = services
.rooms
.state_cache
@@ -129,43 +96,41 @@ pub async fn leave_room(
.is_knocked(user_id, room_id)
.eq(&false);
// Ask a remote server if we don't have this room and are not knocking on it
if dont_have_room.and(not_knocked).await {
if let Err(e) = remote_leave_room(services, user_id, room_id, reason.clone())
.boxed()
.await
{
warn!(%user_id, "Failed to leave room {room_id} remotely: {e}");
// Don't tell the client about this error
}
pin_mut!(is_banned, is_disabled);
let last_state = services
.rooms
.state_cache
.invite_state(user_id, room_id)
.or_else(|_| services.rooms.state_cache.knock_state(user_id, room_id))
.or_else(|_| services.rooms.state_cache.left_state(user_id, room_id))
.await
.ok();
/*
there are three possible cases when leaving a room:
1. the room is banned or disabled, so we're not federating with it.
2. nobody on the homeserver is in the room, which can happen if the user is rejecting an invite
to a room that we don't have any members in.
3. someone else on the homeserver is in the room. in this case we can leave like normal by sending a PDU over federation.
// We always drop the invite, we can't rely on other servers
services
.rooms
.state_cache
.update_membership(
room_id,
user_id,
default_member_content,
user_id,
last_state,
None,
true,
)
.await?;
in cases 1 and 2, we have to update the state cache using `mark_as_left` directly.
otherwise `build_and_append_pdu` will take care of updating the state cache for us.
*/
// `leave_pdu` is the outlier `m.room.member` event which will be synced to the
// user. if it's None the sync handler will create a dummy PDU.
let leave_pdu = if is_banned.or(is_disabled).await {
// case 1: the room is banned/disabled. we don't want to federate with another
// server to leave, so we can't create an outlier PDU.
None
} else if dont_have_room.and(not_knocked).await {
// case 2: ask a remote server to assist us with leaving
// we always mark the room as left locally, regardless of if the federated leave
// failed
remote_leave_room(services, user_id, room_id, reason.clone())
.await
.inspect_err(|err| {
warn!(%user_id, "Failed to leave room {room_id} remotely: {err}");
})
.ok()
} else {
// case 3: we can leave by sending a PDU.
let state_lock = services.rooms.state.mutex.lock(room_id).await;
let Ok(event) = services
let user_member_event_content = services
.rooms
.state_accessor
.room_state_get_content::<RoomMemberEventContent>(
@@ -173,44 +138,61 @@ pub async fn leave_room(
&StateEventType::RoomMember,
user_id.as_str(),
)
.await
else {
debug_warn!(
"Trying to leave a room you are not a member of, marking room as left locally."
);
.await;
return services
.rooms
.state_cache
.update_membership(
room_id,
user_id,
default_member_content,
user_id,
None,
None,
true,
)
.await;
};
match user_member_event_content {
| Ok(content) => {
services
.rooms
.timeline
.build_and_append_pdu(
PduBuilder::state(user_id.to_string(), &RoomMemberEventContent {
membership: MembershipState::Leave,
reason,
join_authorized_via_users_server: None,
is_direct: None,
..content
}),
user_id,
Some(room_id),
&state_lock,
)
.await?;
services
.rooms
.timeline
.build_and_append_pdu(
PduBuilder::state(user_id.to_string(), &RoomMemberEventContent {
membership: MembershipState::Leave,
reason,
join_authorized_via_users_server: None,
is_direct: None,
..event
}),
user_id,
Some(room_id),
&state_lock,
)
.await?;
}
// `build_and_append_pdu` calls `mark_as_left` internally, so we return early.
return Ok(());
},
| Err(_) => {
// an exception to case 3 is if the user isn't even in the room they're trying
// to leave. this can happen if the client's caching is wrong.
debug_warn!(
"Trying to leave a room you are not a member of, marking room as left \
locally."
);
// return the existing leave state, if one exists. `mark_as_left` will then
// update the `roomuserid_leftcount` table, making the leave come down sync
// again.
services
.rooms
.state_cache
.left_state(user_id, room_id)
.await?
},
}
};
services
.rooms
.state_cache
.mark_as_left(user_id, room_id, leave_pdu)
.await;
services
.rooms
.state_cache
.update_joined_count(room_id)
.await;
Ok(())
}
@@ -220,7 +202,7 @@ pub async fn remote_leave_room(
user_id: &UserId,
room_id: &RoomId,
reason: Option<String>,
) -> Result<()> {
) -> Result<Pdu> {
let mut make_leave_response_and_server =
Err!(BadServerResponse("No remote server available to assist in leaving {room_id}."));
@@ -373,7 +355,7 @@ pub async fn remote_leave_room(
&remote_server,
federation::membership::create_leave_event::v2::Request {
room_id: room_id.to_owned(),
event_id,
event_id: event_id.clone(),
pdu: services
.sending
.convert_to_outgoing_federation_event(leave_event.clone())
@@ -382,5 +364,14 @@ pub async fn remote_leave_room(
)
.await?;
Ok(())
services
.rooms
.outlier
.add_pdu_outlier(&event_id, &leave_event);
let leave_pdu = Pdu::from_id_val(&event_id, leave_event).map_err(|e| {
err!(BadServerResponse("Invalid leave PDU received during federated leave: {e:?}"))
})?;
Ok(leave_pdu)
}
+7 -7
View File
@@ -16,7 +16,7 @@
Services,
rooms::{
lazy_loading,
lazy_loading::{Options, Witness},
lazy_loading::{MemberSet, Options},
timeline::PdusIterItem,
},
};
@@ -162,7 +162,7 @@ pub(crate) async fn get_message_events_route(
let state = witness
.map(Option::into_iter)
.map(|option| option.flat_map(Witness::into_iter))
.map(|option| option.flat_map(MemberSet::into_iter))
.map(IterStream::stream)
.into_stream()
.flatten()
@@ -192,7 +192,7 @@ pub(crate) async fn lazy_loading_witness<'a, I>(
services: &Services,
lazy_loading_context: &lazy_loading::Context<'_>,
events: I,
) -> Witness
) -> MemberSet
where
I: Iterator<Item = &'a PdusIterItem> + Clone + Send,
{
@@ -213,10 +213,10 @@ pub(crate) async fn lazy_loading_witness<'a, I>(
let receipts = services
.rooms
.read_receipt
.readreceipts_since(lazy_loading_context.room_id, oldest.into_unsigned());
.readreceipts_since(lazy_loading_context.room_id, Some(oldest.into_unsigned()));
pin_mut!(receipts);
let witness: Witness = events
let witness: MemberSet = events
.stream()
.map(ref_at!(1))
.map(Event::sender)
@@ -224,7 +224,7 @@ pub(crate) async fn lazy_loading_witness<'a, I>(
.chain(
receipts
.ready_take_while(|(_, c, _)| *c <= newest.into_unsigned())
.map(|(user_id, ..)| user_id.to_owned()),
.map(|(user_id, ..)| user_id),
)
.collect()
.await;
@@ -232,7 +232,7 @@ pub(crate) async fn lazy_loading_witness<'a, I>(
services
.rooms
.lazy_loading
.witness_retain(witness, lazy_loading_context)
.retain_lazy_members(witness, lazy_loading_context)
.await
}
+4 -3
View File
@@ -97,11 +97,12 @@ pub(crate) async fn upgrade_room_route(
// Create a replacement room
let room_features = RoomVersion::new(&body.new_version)?;
let replacement_room: Option<&RoomId> = if room_features.room_ids_as_hashes {
None
let replacement_room_owned = if !room_features.room_ids_as_hashes {
Some(RoomId::new(services.globals.server_name()))
} else {
Some(&RoomId::new(services.globals.server_name()))
None
};
let replacement_room: Option<&RoomId> = replacement_room_owned.as_ref().map(AsRef::as_ref);
let replacement_room_tmp = match replacement_room {
| Some(v) => v,
| None => &RoomId::new(services.globals.server_name()),
+1 -1
View File
@@ -44,7 +44,7 @@ pub(crate) async fn get_hierarchy_route(
.as_ref()
.and_then(|s| PaginationToken::from_str(s).ok());
// Should prevent unexpeded behaviour in (bad) clients
// Should prevent unexpected behaviour in (bad) clients
if let Some(ref token) = key {
if token.suggested_only != body.suggested_only || token.max_depth != max_depth {
return Err!(Request(InvalidParam(
+91 -33
View File
@@ -1,65 +1,123 @@
mod v3;
mod v4;
mod v5;
use std::collections::VecDeque;
use conduwuit::{
Error, PduCount, Result,
Event, PduCount, Result, err,
matrix::pdu::PduEvent,
ref_at, trace,
utils::stream::{BroadbandExt, ReadyExt, TryIgnore},
};
use conduwuit_service::Services;
use futures::{StreamExt, pin_mut};
use futures::StreamExt;
use ruma::{
RoomId, UserId,
OwnedUserId, RoomId, UserId,
events::TimelineEventType::{
self, Beacon, CallInvite, PollStart, RoomEncrypted, RoomMessage, Sticker,
},
};
pub(crate) use self::{
v3::sync_events_route, v4::sync_events_v4_route, v5::sync_events_v5_route,
};
pub(crate) use self::{v3::sync_events_route, v5::sync_events_v5_route};
pub(crate) const DEFAULT_BUMP_TYPES: &[TimelineEventType; 6] =
&[CallInvite, PollStart, Beacon, RoomEncrypted, RoomMessage, Sticker];
#[derive(Default)]
pub(crate) struct TimelinePdus {
pub pdus: VecDeque<(PduCount, PduEvent)>,
pub limited: bool,
}
impl TimelinePdus {
fn senders(&self) -> impl Iterator<Item = OwnedUserId> {
self.pdus
.iter()
.map(ref_at!(1))
.map(Event::sender)
.map(Into::into)
}
}
/// Load up to `limit` PDUs in the range (starting_count, ending_count].
async fn load_timeline(
services: &Services,
sender_user: &UserId,
room_id: &RoomId,
roomsincecount: PduCount,
next_batch: Option<PduCount>,
starting_count: Option<PduCount>,
ending_count: Option<PduCount>,
limit: usize,
) -> Result<(Vec<(PduCount, PduEvent)>, bool), Error> {
let last_timeline_count = services
.rooms
.timeline
.last_timeline_count(Some(sender_user), room_id)
.await?;
) -> Result<TimelinePdus> {
let mut pdu_stream = match starting_count {
| Some(starting_count) => {
let last_timeline_count = services
.rooms
.timeline
.last_timeline_count(Some(sender_user), room_id)
.await
.map_err(|err| {
err!(Database(warn!("Failed to fetch end of room timeline: {}", err)))
})?;
if last_timeline_count <= roomsincecount {
return Ok((Vec::new(), false));
}
if last_timeline_count <= starting_count {
// no messages have been sent in this room since `starting_count`
return Ok(TimelinePdus::default());
}
let non_timeline_pdus = services
.rooms
.timeline
.pdus_rev(Some(sender_user), room_id, None)
.ignore_err()
.ready_skip_while(|&(pducount, _)| pducount > next_batch.unwrap_or_else(PduCount::max))
.ready_take_while(|&(pducount, _)| pducount > roomsincecount);
// for incremental sync, stream from the DB all PDUs which were sent after
// `starting_count` but before `ending_count`, including `ending_count` but
// not `starting_count`. this code is pretty similar to the initial sync
// branch, they're separate to allow for future optimization
services
.rooms
.timeline
.pdus_rev(
Some(sender_user),
room_id,
ending_count.map(|count| count.saturating_add(1)),
)
.ignore_err()
.ready_take_while(move |&(pducount, _)| pducount > starting_count)
.boxed()
},
| None => {
// For initial sync, stream from the DB all PDUs before and including
// `ending_count` in reverse order
services
.rooms
.timeline
.pdus_rev(
Some(sender_user),
room_id,
ending_count.map(|count| count.saturating_add(1)),
)
.ignore_err()
.boxed()
},
};
// Take the last events for the timeline
pin_mut!(non_timeline_pdus);
let timeline_pdus: Vec<_> = non_timeline_pdus.by_ref().take(limit).collect().await;
// Return at most `limit` PDUs from the stream
let pdus = pdu_stream
.by_ref()
.take(limit)
.ready_fold(VecDeque::with_capacity(limit), |mut pdus, item| {
pdus.push_front(item);
pdus
})
.await;
let timeline_pdus: Vec<_> = timeline_pdus.into_iter().rev().collect();
// The timeline is limited if there are still more PDUs in the stream
let limited = pdu_stream.next().await.is_some();
// They /sync response doesn't always return all messages, so we say the output
// is limited unless there are events in non_timeline_pdus
let limited = non_timeline_pdus.next().await.is_some();
trace!(
"syncing {:?} timeline pdus from {:?} to {:?} (limited = {:?})",
pdus.len(),
starting_count,
ending_count,
limited,
);
Ok((timeline_pdus, limited))
Ok(TimelinePdus { pdus, limited })
}
async fn share_encrypted_room(
File diff suppressed because it is too large Load Diff
+852
View File
@@ -0,0 +1,852 @@
use std::collections::{BTreeMap, HashSet};
use conduwuit::{
Result, at, debug_warn, err, extract_variant,
matrix::{
Event,
pdu::{PduCount, PduEvent},
},
trace,
utils::{
BoolExt, IterStream, ReadyExt, TryFutureExtExt,
math::ruma_from_u64,
stream::{TryIgnore, WidebandExt},
},
warn,
};
use conduwuit_service::Services;
use futures::{
FutureExt, StreamExt, TryFutureExt,
future::{OptionFuture, join, join3, join4, try_join, try_join3},
};
use ruma::{
OwnedRoomId, OwnedUserId, RoomId, UserId,
api::client::sync::sync_events::{
UnreadNotificationsCount,
v3::{Ephemeral, JoinedRoom, RoomAccountData, RoomSummary, State as RoomState, Timeline},
},
events::{
AnyRawAccountDataEvent, StateEventType,
TimelineEventType::*,
room::member::{MembershipState, RoomMemberEventContent},
},
serde::Raw,
uint,
};
use service::rooms::short::ShortStateHash;
use super::{load_timeline, share_encrypted_room};
use crate::client::{
TimelinePdus, ignored_filter,
sync::v3::{
DEFAULT_TIMELINE_LIMIT, DeviceListUpdates, SyncContext, prepare_lazily_loaded_members,
state::{build_state_incremental, build_state_initial},
},
};
/// Generate the sync response for a room the user is joined to.
#[tracing::instrument(
name = "joined",
level = "debug",
skip_all,
fields(
room_id = ?room_id,
syncing_user = ?sync_context.syncing_user,
),
)]
pub(super) async fn load_joined_room(
services: &Services,
sync_context: SyncContext<'_>,
ref room_id: OwnedRoomId,
) -> Result<(JoinedRoom, DeviceListUpdates)> {
/*
Building a sync response involves many steps which all depend on each other.
To parallelize the process as much as possible, each step is divided into its own function,
and `join*` functions are used to perform steps in parallel which do not depend on each other.
*/
let (
account_data,
ephemeral,
StateAndTimeline {
state_events,
timeline,
summary,
notification_counts,
device_list_updates,
},
) = try_join3(
build_account_data(services, sync_context, room_id),
build_ephemeral(services, sync_context, room_id),
build_state_and_timeline(services, sync_context, room_id),
)
.boxed()
.await?;
if !timeline.is_empty() || !state_events.is_empty() {
trace!(
"syncing {} timeline events (limited = {}) and {} state events",
timeline.events.len(),
timeline.limited,
state_events.len()
);
}
let joined_room = JoinedRoom {
account_data,
summary: summary.unwrap_or_default(),
unread_notifications: notification_counts.unwrap_or_default(),
timeline,
state: RoomState {
events: state_events.into_iter().map(Event::into_format).collect(),
},
ephemeral,
unread_thread_notifications: BTreeMap::new(),
};
Ok((joined_room, device_list_updates))
}
/// Collect changes to the syncing user's account data events.
#[tracing::instrument(level = "debug", skip_all)]
async fn build_account_data(
services: &Services,
SyncContext {
syncing_user,
last_sync_end_count,
current_count,
..
}: SyncContext<'_>,
room_id: &RoomId,
) -> Result<RoomAccountData> {
let account_data_changes = services
.account_data
.changes_since(Some(room_id), syncing_user, last_sync_end_count, Some(current_count))
.ready_filter_map(|e| extract_variant!(e, AnyRawAccountDataEvent::Room))
.collect()
.await;
Ok(RoomAccountData { events: account_data_changes })
}
/// Collect new ephemeral events.
#[tracing::instrument(level = "debug", skip_all)]
async fn build_ephemeral(
services: &Services,
SyncContext { syncing_user, last_sync_end_count, .. }: SyncContext<'_>,
room_id: &RoomId,
) -> Result<Ephemeral> {
// note: some of the futures below are boxed. this is because, without the box,
// rustc produces over thirty inscrutable errors in `mod.rs` at the call-site
// of `load_joined_room`. I don't know why boxing them fixes this -- it seems
// to be related to the async closures and borrowing from the sync context.
// collect updates to read receipts
let receipt_events = services
.rooms
.read_receipt
.readreceipts_since(room_id, last_sync_end_count)
.filter_map(async |(read_user, _, edu)| {
let is_ignored = services
.users
.user_is_ignored(&read_user, syncing_user)
.await;
// filter out read receipts for ignored users
is_ignored.or_some(edu)
})
.collect::<Vec<_>>()
.boxed();
// collect the updated list of typing users, if it's changed
let typing_event = async {
let should_send_typing_event = match last_sync_end_count {
| Some(last_sync_end_count) => {
match services.rooms.typing.last_typing_update(room_id).await {
| Ok(last_typing_update) => {
// update the typing list if the users typing have changed since the last
// sync
last_typing_update > last_sync_end_count
},
| Err(err) => {
warn!("Error checking last typing update: {}", err);
return None;
},
}
},
// always update the typing list on an initial sync
| None => true,
};
if should_send_typing_event {
let event = services
.rooms
.typing
.typings_event_for_user(room_id, syncing_user)
.await;
if let Ok(event) = event {
return Some(
Raw::new(&event)
.expect("typing event should be valid")
.cast(),
);
}
}
None
};
// collect the syncing user's private-read marker, if it's changed
let private_read_event = async {
let should_send_private_read = match last_sync_end_count {
| Some(last_sync_end_count) => {
let last_privateread_update = services
.rooms
.read_receipt
.last_privateread_update(syncing_user, room_id)
.await;
// update the marker if it's changed since the last sync
last_privateread_update > last_sync_end_count
},
// always update the marker on an initial sync
| None => true,
};
if should_send_private_read {
services
.rooms
.read_receipt
.private_read_get(room_id, syncing_user)
.await
.ok()
} else {
None
}
};
let (receipt_events, typing_event, private_read_event) =
join3(receipt_events, typing_event, private_read_event).await;
let mut edus = receipt_events;
edus.extend(typing_event);
edus.extend(private_read_event);
Ok(Ephemeral { events: edus })
}
/// A struct to hold the state events, timeline, and other data which is
/// computed from them.
struct StateAndTimeline {
state_events: Vec<PduEvent>,
timeline: Timeline,
summary: Option<RoomSummary>,
notification_counts: Option<UnreadNotificationsCount>,
device_list_updates: DeviceListUpdates,
}
/// Compute changes to the room's state and timeline.
#[tracing::instrument(level = "debug", skip_all)]
async fn build_state_and_timeline(
services: &Services,
sync_context: SyncContext<'_>,
room_id: &RoomId,
) -> Result<StateAndTimeline> {
let (shortstatehashes, timeline) = try_join(
fetch_shortstatehashes(services, sync_context, room_id),
build_timeline(services, sync_context, room_id),
)
.await?;
let (state_events, notification_counts, joined_since_last_sync) = try_join3(
build_state_events(services, sync_context, room_id, shortstatehashes, &timeline),
build_notification_counts(services, sync_context, room_id, &timeline),
check_joined_since_last_sync(services, shortstatehashes, sync_context),
)
.await?;
// the timeline should always include at least one PDU if the syncing user
// joined since the last sync, that being the syncing user's join event. if
// it's empty something is wrong.
if joined_since_last_sync && timeline.pdus.is_empty() {
warn!("timeline for newly joined room is empty");
}
let (summary, device_list_updates) = try_join(
build_room_summary(
services,
sync_context,
room_id,
shortstatehashes,
&timeline,
&state_events,
joined_since_last_sync,
),
build_device_list_updates(
services,
sync_context,
room_id,
shortstatehashes,
&state_events,
joined_since_last_sync,
),
)
.await?;
// the token which may be passed to the messages endpoint to backfill room
// history
let prev_batch = timeline.pdus.front().map(at!(0));
// note: we always indicate a limited timeline if the syncing user just joined
// the room, to indicate to the client that it should request backfill (and to
// copy Synapse's behavior). for federated room joins, the `timeline` will
// usually only include the syncing user's join event.
let limited = timeline.limited || joined_since_last_sync;
// filter out ignored events from the timeline and convert the PDUs into Ruma's
// AnySyncTimelineEvent type
let filtered_timeline = timeline
.pdus
.into_iter()
.stream()
.wide_filter_map(|item| ignored_filter(services, item, sync_context.syncing_user))
.map(at!(1))
.map(Event::into_format)
.collect::<Vec<_>>()
.await;
Ok(StateAndTimeline {
state_events,
timeline: Timeline {
limited,
prev_batch: prev_batch.as_ref().map(ToString::to_string),
events: filtered_timeline,
},
summary,
notification_counts,
device_list_updates,
})
}
/// Shortstatehashes necessary to compute what state events to sync.
#[derive(Clone, Copy)]
struct ShortStateHashes {
/// The current state of the syncing room.
current_shortstatehash: ShortStateHash,
/// The state of the syncing room at the end of the last sync.
last_sync_end_shortstatehash: Option<ShortStateHash>,
}
/// Fetch the current_shortstatehash and last_sync_end_shortstatehash.
#[tracing::instrument(level = "debug", skip_all)]
async fn fetch_shortstatehashes(
services: &Services,
SyncContext { last_sync_end_count, current_count, .. }: SyncContext<'_>,
room_id: &RoomId,
) -> Result<ShortStateHashes> {
// the room state currently.
// TODO: this should be the room state as of `current_count`, but there's no way
// to get that right now.
let current_shortstatehash = services
.rooms
.state
.get_room_shortstatehash(room_id)
.map_err(|_| err!(Database(error!("Room {room_id} has no state"))));
// the room state as of the end of the last sync.
// this will be None if we are doing an initial sync or if we just joined this
// room.
let last_sync_end_shortstatehash =
OptionFuture::from(last_sync_end_count.map(|last_sync_end_count| {
// look up the shortstatehash saved by the last sync's call to
// `associate_token_shortstatehash`
services
.rooms
.user
.get_token_shortstatehash(room_id, last_sync_end_count)
.inspect_err(move |_| {
debug_warn!(
token = last_sync_end_count,
"Room has no shortstatehash for this token"
);
})
.ok()
}))
.map(Option::flatten)
.map(Ok);
let (current_shortstatehash, last_sync_end_shortstatehash) =
try_join(current_shortstatehash, last_sync_end_shortstatehash).await?;
/*
associate the `current_count` with the `current_shortstatehash`, so we can
use it on the next sync as the `last_sync_end_shortstatehash`.
TODO: the table written to by this call grows extremely fast, gaining one new entry for each
joined room on _every single sync request_. we need to find a better way to remember the shortstatehash
between syncs.
*/
services
.rooms
.user
.associate_token_shortstatehash(room_id, current_count, current_shortstatehash)
.await;
Ok(ShortStateHashes {
current_shortstatehash,
last_sync_end_shortstatehash,
})
}
/// Fetch recent timeline events.
#[tracing::instrument(level = "debug", skip_all)]
async fn build_timeline(
services: &Services,
sync_context: SyncContext<'_>,
room_id: &RoomId,
) -> Result<TimelinePdus> {
let SyncContext {
syncing_user,
last_sync_end_count,
current_count,
filter,
..
} = sync_context;
/*
determine the maximum number of events to return in this sync.
if the sync filter specifies a limit, that will be used, otherwise
`DEFAULT_TIMELINE_LIMIT` will be used. `DEFAULT_TIMELINE_LIMIT` will also be
used if the limit is somehow greater than usize::MAX.
*/
let timeline_limit = filter
.room
.timeline
.limit
.and_then(|limit| limit.try_into().ok())
.unwrap_or(DEFAULT_TIMELINE_LIMIT);
load_timeline(
services,
syncing_user,
room_id,
last_sync_end_count.map(PduCount::Normal),
Some(PduCount::Normal(current_count)),
timeline_limit,
)
.await
}
/// Calculate the state events to sync.
async fn build_state_events(
services: &Services,
sync_context: SyncContext<'_>,
room_id: &RoomId,
shortstatehashes: ShortStateHashes,
timeline: &TimelinePdus,
) -> Result<Vec<PduEvent>> {
let SyncContext {
syncing_user,
last_sync_end_count,
full_state,
..
} = sync_context;
let ShortStateHashes {
current_shortstatehash,
last_sync_end_shortstatehash,
} = shortstatehashes;
// the spec states that the `state` property only includes state events up to
// the beginning of the timeline, so we determine the state of the syncing room
// as of the first timeline event. NOTE: this explanation is not entirely
// accurate; see the implementation of `build_state_incremental`.
let timeline_start_shortstatehash = async {
if let Some((_, pdu)) = timeline.pdus.front() {
if let Ok(shortstatehash) = services
.rooms
.state_accessor
.pdu_shortstatehash(&pdu.event_id)
.await
{
return shortstatehash;
}
}
current_shortstatehash
};
// the user IDs of members whose membership needs to be sent to the client, if
// lazy-loading is enabled.
let lazily_loaded_members =
prepare_lazily_loaded_members(services, sync_context, room_id, timeline.senders());
let (timeline_start_shortstatehash, lazily_loaded_members) =
join(timeline_start_shortstatehash, lazily_loaded_members).await;
// compute the state delta between the previous sync and this sync.
match (last_sync_end_count, last_sync_end_shortstatehash) {
/*
if `last_sync_end_count` is Some (meaning this is an incremental sync), and `last_sync_end_shortstatehash`
is Some (meaning the syncing user didn't just join this room for the first time ever), and `full_state` is false,
then use `build_state_incremental`.
*/
| (Some(last_sync_end_count), Some(last_sync_end_shortstatehash)) if !full_state =>
build_state_incremental(
services,
syncing_user,
room_id,
PduCount::Normal(last_sync_end_count),
last_sync_end_shortstatehash,
timeline_start_shortstatehash,
current_shortstatehash,
timeline,
lazily_loaded_members.as_ref(),
)
.boxed()
.await,
/*
otherwise use `build_state_initial`. note that this branch will be taken if the user joined this room since the last sync
for the first time ever, because in that case we have no `last_sync_end_shortstatehash` and can't correctly calculate
the state using the incremental sync algorithm.
*/
| _ =>
build_state_initial(
services,
syncing_user,
timeline_start_shortstatehash,
lazily_loaded_members.as_ref(),
)
.boxed()
.await,
}
}
/// Compute the number of unread notifications in this room.
#[tracing::instrument(level = "debug", skip_all)]
async fn build_notification_counts(
services: &Services,
SyncContext { syncing_user, last_sync_end_count, .. }: SyncContext<'_>,
room_id: &RoomId,
timeline: &TimelinePdus,
) -> Result<Option<UnreadNotificationsCount>> {
// determine whether to actually update the notification counts
let should_send_notification_counts = async {
// if we're going to sync some timeline events, the notification count has
// definitely changed to include them
if !timeline.pdus.is_empty() {
return true;
}
// if this is an initial sync, we need to send notification counts because the
// client doesn't know what they are yet
let Some(last_sync_end_count) = last_sync_end_count else {
return true;
};
let last_notification_read = services
.rooms
.user
.last_notification_read(syncing_user, room_id)
.await;
// if the syncing user has read the events we sent during the last sync, we need
// to send a new notification count on this sync.
if last_notification_read > last_sync_end_count {
return true;
}
// otherwise, nothing's changed.
false
};
if should_send_notification_counts.await {
let (notification_count, highlight_count) = join(
services
.rooms
.user
.notification_count(syncing_user, room_id)
.map(TryInto::try_into)
.unwrap_or(uint!(0)),
services
.rooms
.user
.highlight_count(syncing_user, room_id)
.map(TryInto::try_into)
.unwrap_or(uint!(0)),
)
.await;
trace!(?notification_count, ?highlight_count, "syncing new notification counts");
Ok(Some(UnreadNotificationsCount {
notification_count: Some(notification_count),
highlight_count: Some(highlight_count),
}))
} else {
Ok(None)
}
}
/// Check if the syncing user joined the room since their last incremental sync.
#[tracing::instrument(level = "debug", skip_all)]
async fn check_joined_since_last_sync(
services: &Services,
ShortStateHashes { last_sync_end_shortstatehash, .. }: ShortStateHashes,
SyncContext { syncing_user, .. }: SyncContext<'_>,
) -> Result<bool> {
// fetch the syncing user's membership event during the last sync.
// this will be None if `previous_sync_end_shortstatehash` is None.
let membership_during_previous_sync = match last_sync_end_shortstatehash {
| Some(last_sync_end_shortstatehash) => services
.rooms
.state_accessor
.state_get_content(
last_sync_end_shortstatehash,
&StateEventType::RoomMember,
syncing_user.as_str(),
)
.await
.inspect_err(|_| debug_warn!("User has no previous membership"))
.ok(),
| None => None,
};
// TODO: If the requesting user got state-reset out of the room, this
// will be `true` when it shouldn't be. this function should never be called
// in that situation, but it may be if the membership cache didn't get updated.
// the root cause of this needs to be addressed
let joined_since_last_sync =
membership_during_previous_sync.is_none_or(|content: RoomMemberEventContent| {
content.membership != MembershipState::Join
});
if joined_since_last_sync {
trace!("user joined since last sync");
}
Ok(joined_since_last_sync)
}
/// Build the `summary` field of the room object, which includes
/// the number of joined and invited users and the room's heroes.
#[tracing::instrument(level = "debug", skip_all)]
async fn build_room_summary(
services: &Services,
SyncContext { syncing_user, .. }: SyncContext<'_>,
room_id: &RoomId,
ShortStateHashes { current_shortstatehash, .. }: ShortStateHashes,
timeline: &TimelinePdus,
state_events: &[PduEvent],
joined_since_last_sync: bool,
) -> Result<Option<RoomSummary>> {
// determine whether any events in the state or timeline are membership events.
let are_syncing_membership_events = timeline
.pdus
.iter()
.map(|(_, pdu)| pdu)
.chain(state_events.iter())
.any(|event| event.kind == RoomMember);
/*
we only need to send an updated room summary if:
1. there are membership events in the state or timeline, because they might have changed the
membership counts or heroes, or
2. the syncing user just joined this room, which usually implies #1 because their join event should be in the timeline.
*/
if !(are_syncing_membership_events || joined_since_last_sync) {
return Ok(None);
}
let joined_member_count = services
.rooms
.state_cache
.room_joined_count(room_id)
.unwrap_or(0);
let invited_member_count = services
.rooms
.state_cache
.room_invited_count(room_id)
.unwrap_or(0);
let has_name = services
.rooms
.state_accessor
.state_contains_type(current_shortstatehash, &StateEventType::RoomName);
let has_canonical_alias = services
.rooms
.state_accessor
.state_contains_type(current_shortstatehash, &StateEventType::RoomCanonicalAlias);
let (joined_member_count, invited_member_count, has_name, has_canonical_alias) =
join4(joined_member_count, invited_member_count, has_name, has_canonical_alias).await;
// only send heroes if the room has neither a name nor a canonical alias
let heroes = if !(has_name || has_canonical_alias) {
Some(build_heroes(services, room_id, syncing_user, current_shortstatehash).await)
} else {
None
};
trace!(
?joined_member_count,
?invited_member_count,
heroes_length = heroes.as_ref().map(HashSet::len),
"syncing updated summary"
);
Ok(Some(RoomSummary {
heroes: heroes
.map(|heroes| heroes.into_iter().collect())
.unwrap_or_default(),
joined_member_count: Some(ruma_from_u64(joined_member_count)),
invited_member_count: Some(ruma_from_u64(invited_member_count)),
}))
}
/// Fetch the user IDs to include in the `m.heroes` property of the room
/// summary.
async fn build_heroes(
services: &Services,
room_id: &RoomId,
syncing_user: &UserId,
current_shortstatehash: ShortStateHash,
) -> HashSet<OwnedUserId> {
const MAX_HERO_COUNT: usize = 5;
// fetch joined members from the state cache first
let joined_members_stream = services
.rooms
.state_cache
.room_members(room_id)
.map(ToOwned::to_owned);
// then fetch invited members
let invited_members_stream = services
.rooms
.state_cache
.room_members_invited(room_id)
.map(ToOwned::to_owned);
// then as a last resort fetch every membership event
let all_members_stream = services
.rooms
.short
.multi_get_statekey_from_short(
services
.rooms
.state_accessor
.state_full_shortids(current_shortstatehash)
.ignore_err()
.ready_filter_map(|(key, _)| Some(key)),
)
.ignore_err()
.ready_filter_map(|(event_type, state_key)| {
if event_type == StateEventType::RoomMember {
state_key.to_string().try_into().ok()
} else {
None
}
});
joined_members_stream
.chain(invited_members_stream)
.chain(all_members_stream)
// the hero list should never include the syncing user
.ready_filter(|user_id| user_id != syncing_user)
.take(MAX_HERO_COUNT)
.collect()
.await
}
/// Collect updates to users' device lists for E2EE.
#[tracing::instrument(level = "debug", skip_all)]
async fn build_device_list_updates(
services: &Services,
SyncContext {
syncing_user,
last_sync_end_count,
current_count,
..
}: SyncContext<'_>,
room_id: &RoomId,
ShortStateHashes { current_shortstatehash, .. }: ShortStateHashes,
state_events: &Vec<PduEvent>,
joined_since_last_sync: bool,
) -> Result<DeviceListUpdates> {
let is_encrypted_room = services
.rooms
.state_accessor
.state_get(current_shortstatehash, &StateEventType::RoomEncryption, "")
.is_ok();
// initial syncs don't include device updates, and rooms which aren't encrypted
// don't affect them, so return early in either of those cases
if last_sync_end_count.is_none() || !(is_encrypted_room.await) {
return Ok(DeviceListUpdates::new());
}
let mut device_list_updates = DeviceListUpdates::new();
// add users with changed keys to the `changed` list
services
.users
.room_keys_changed(room_id, last_sync_end_count, Some(current_count))
.map(at!(0))
.map(ToOwned::to_owned)
.ready_for_each(|user_id| {
device_list_updates.changed.insert(user_id);
})
.await;
// add users who now share encrypted rooms to `changed` and
// users who no longer share encrypted rooms to `left`
for state_event in state_events {
if state_event.kind == RoomMember {
let Some(content): Option<RoomMemberEventContent> = state_event.get_content().ok()
else {
continue;
};
let Some(user_id): Option<OwnedUserId> = state_event
.state_key
.as_ref()
.and_then(|key| key.parse().ok())
else {
continue;
};
{
use MembershipState::*;
if matches!(content.membership, Leave | Join) {
let shares_encrypted_room =
share_encrypted_room(services, syncing_user, &user_id, Some(room_id))
.await;
match content.membership {
| Leave if !shares_encrypted_room => {
device_list_updates.left.insert(user_id);
},
| Join if joined_since_last_sync || shares_encrypted_room => {
device_list_updates.changed.insert(user_id);
},
| _ => (),
}
}
}
}
}
if !device_list_updates.is_empty() {
trace!(
changed = device_list_updates.changed.len(),
left = device_list_updates.left.len(),
"syncing device list updates"
);
}
Ok(device_list_updates)
}
+349
View File
@@ -0,0 +1,349 @@
use conduwuit::{
Event, PduCount, PduEvent, Result, at, debug_warn,
pdu::EventHash,
trace,
utils::{self, IterStream, future::ReadyEqExt, stream::WidebandExt as _},
};
use futures::{StreamExt, future::join};
use ruma::{
EventId, OwnedRoomId, RoomId,
api::client::sync::sync_events::v3::{LeftRoom, RoomAccountData, State, Timeline},
events::{StateEventType, TimelineEventType},
uint,
};
use serde_json::value::RawValue;
use service::{Services, rooms::short::ShortStateHash};
use crate::client::{
TimelinePdus, ignored_filter,
sync::{
load_timeline,
v3::{
DEFAULT_TIMELINE_LIMIT, SyncContext, prepare_lazily_loaded_members,
state::build_state_initial,
},
},
};
#[tracing::instrument(
name = "left",
level = "debug",
skip_all,
fields(
room_id = %room_id,
),
)]
#[allow(clippy::too_many_arguments)]
pub(super) async fn load_left_room(
services: &Services,
sync_context: SyncContext<'_>,
ref room_id: OwnedRoomId,
leave_membership_event: Option<PduEvent>,
) -> Result<Option<LeftRoom>> {
let SyncContext {
syncing_user,
last_sync_end_count,
current_count,
filter,
..
} = sync_context;
// the global count as of the moment the user left the room
let Some(left_count) = services
.rooms
.state_cache
.get_left_count(room_id, syncing_user)
.await
.ok()
else {
// if we get here, the membership cache is incorrect, likely due to a state
// reset
debug_warn!("attempting to sync left room but no left count exists");
return Ok(None);
};
// return early if we haven't gotten to this leave yet.
// this can happen if the user leaves while a sync response is being generated
if current_count < left_count {
return Ok(None);
}
// return early if this is an incremental sync, and we've already synced this
// leave to the user, and `include_leave` isn't set on the filter.
if !filter.room.include_leave && last_sync_end_count >= Some(left_count) {
return Ok(None);
}
if let Some(ref leave_membership_event) = leave_membership_event {
debug_assert_eq!(
leave_membership_event.kind,
TimelineEventType::RoomMember,
"leave PDU should be m.room.member"
);
}
let does_not_exist = services.rooms.metadata.exists(room_id).eq(&false).await;
let (timeline, state_events) = match leave_membership_event {
| Some(leave_membership_event) if does_not_exist => {
/*
we have none PDUs with left beef for this room, likely because it was a rejected invite to a room
which nobody on this homeserver is in. `leave_pdu` is the remote-assisted outlier leave event for the room,
which is all we can send to the client.
if this is an initial sync, don't include this room at all to keep the client from asking for
state that we don't have.
*/
if last_sync_end_count.is_none() {
return Ok(None);
}
trace!("syncing remote-assisted leave PDU");
(TimelinePdus::default(), vec![leave_membership_event])
},
| Some(leave_membership_event) => {
// we have this room in our DB, and can fetch the state and timeline from when
// the user left.
let leave_state_key = syncing_user;
debug_assert_eq!(
Some(leave_state_key.as_str()),
leave_membership_event.state_key(),
"leave PDU should be for the user requesting the sync"
);
// the shortstatehash of the state _immediately before_ the syncing user left
// this room. the state represented here _does not_ include
// `leave_membership_event`.
let leave_shortstatehash = services
.rooms
.state_accessor
.pdu_shortstatehash(&leave_membership_event.event_id)
.await?;
let prev_membership_event = services
.rooms
.state_accessor
.state_get(
leave_shortstatehash,
&StateEventType::RoomMember,
leave_state_key.as_str(),
)
.await?;
build_left_state_and_timeline(
services,
sync_context,
room_id,
leave_membership_event,
leave_shortstatehash,
prev_membership_event,
)
.await?
},
| None => {
/*
no leave event was actually sent in this room, but we still need to pretend
like the user left it. this is usually because the room was banned by a server admin.
if this is an incremental sync, generate a fake leave event to make the room vanish from clients.
otherwise we don't tell the client about this room at all.
*/
if last_sync_end_count.is_none() {
return Ok(None);
}
trace!("syncing dummy leave event");
(TimelinePdus::default(), vec![create_dummy_leave_event(
services,
sync_context,
room_id,
)])
},
};
let raw_timeline_pdus = timeline
.pdus
.into_iter()
.stream()
// filter out ignored events from the timeline
.wide_filter_map(|item| ignored_filter(services, item, syncing_user))
.map(at!(1))
.map(Event::into_format)
.collect::<Vec<_>>()
.await;
Ok(Some(LeftRoom {
account_data: RoomAccountData { events: Vec::new() },
timeline: Timeline {
limited: timeline.limited,
prev_batch: Some(current_count.to_string()),
events: raw_timeline_pdus,
},
state: State {
events: state_events.into_iter().map(Event::into_format).collect(),
},
}))
}
async fn build_left_state_and_timeline(
services: &Services,
sync_context: SyncContext<'_>,
room_id: &RoomId,
leave_membership_event: PduEvent,
leave_shortstatehash: ShortStateHash,
prev_membership_event: PduEvent,
) -> Result<(TimelinePdus, Vec<PduEvent>)> {
let SyncContext {
syncing_user,
last_sync_end_count,
filter,
..
} = sync_context;
let timeline_start_count = if let Some(last_sync_end_count) = last_sync_end_count {
// for incremental syncs, start the timeline after `since`
PduCount::Normal(last_sync_end_count)
} else {
// for initial syncs, start the timeline after the previous membership
// event. we don't want to include the membership event itself
// because clients get confused when they see a `join`
// membership event in a `leave` room.
services
.rooms
.timeline
.get_pdu_count(&prev_membership_event.event_id)
.await?
};
// end the timeline at the user's leave event
let timeline_end_count = services
.rooms
.timeline
.get_pdu_count(leave_membership_event.event_id())
.await?;
// limit the timeline using the same logic as for joined rooms
let timeline_limit = filter
.room
.timeline
.limit
.and_then(|limit| limit.try_into().ok())
.unwrap_or(DEFAULT_TIMELINE_LIMIT);
let timeline = load_timeline(
services,
syncing_user,
room_id,
Some(timeline_start_count),
Some(timeline_end_count),
timeline_limit,
)
.await?;
let timeline_start_shortstatehash = async {
if let Some((_, pdu)) = timeline.pdus.front() {
if let Ok(shortstatehash) = services
.rooms
.state_accessor
.pdu_shortstatehash(&pdu.event_id)
.await
{
return shortstatehash;
}
}
// the timeline generally should not be empty (see the TODO further down),
// but in case it is we use `leave_shortstatehash` as the state to
// send
leave_shortstatehash
};
let lazily_loaded_members =
prepare_lazily_loaded_members(services, sync_context, room_id, timeline.senders());
let (timeline_start_shortstatehash, lazily_loaded_members) =
join(timeline_start_shortstatehash, lazily_loaded_members).await;
// TODO: calculate incremental state for incremental syncs.
// always calculating initial state _works_ but returns more data and does
// more processing than strictly necessary.
let mut state = build_state_initial(
services,
syncing_user,
timeline_start_shortstatehash,
lazily_loaded_members.as_ref(),
)
.await?;
/*
remove membership events for the syncing user from state.
usually, `state` should include a `join` membership event and `timeline` should include a `leave` one.
however, the matrix-js-sdk gets confused when this happens (see [1]) and doesn't process the room leave,
so we have to filter out the membership from `state`.
NOTE: we are sending more information than synapse does in this scenario, because we always
calculate `state` for initial syncs, even when the sync being performed is incremental.
however, the specification does not forbid sending extraneous events in `state`.
TODO: there is an additional bug at play here. sometimes `load_joined_room` syncs the `leave` event
before `load_left_room` does, which means the `timeline` we sync immediately after a leave is empty.
this shouldn't happen -- `timeline` should always include the `leave` event. this is probably
a race condition with the membership state cache.
[1]: https://github.com/matrix-org/matrix-js-sdk/issues/5071
*/
// `state` should only ever include one membership event for the syncing user
let membership_event_index = state.iter().position(|pdu| {
*pdu.event_type() == TimelineEventType::RoomMember
&& pdu.state_key() == Some(syncing_user.as_str())
});
if let Some(index) = membership_event_index {
// the ordering of events in `state` does not matter
state.swap_remove(index);
}
trace!(
?timeline_start_count,
?timeline_end_count,
"syncing {} timeline events (limited = {}) and {} state events",
timeline.pdus.len(),
timeline.limited,
state.len()
);
Ok((timeline, state))
}
fn create_dummy_leave_event(
services: &Services,
SyncContext { syncing_user, .. }: SyncContext<'_>,
room_id: &RoomId,
) -> PduEvent {
// TODO: because this event ID is random, it could cause caching issues with
// clients. perhaps a database table could be created to hold these dummy
// events, or they could be stored as outliers?
PduEvent {
event_id: EventId::new(services.globals.server_name()),
sender: syncing_user.to_owned(),
origin: None,
origin_server_ts: utils::millis_since_unix_epoch()
.try_into()
.expect("Timestamp is valid js_int value"),
kind: TimelineEventType::RoomMember,
content: RawValue::from_string(r#"{"membership": "leave"}"#.to_owned()).unwrap(),
state_key: Some(syncing_user.as_str().into()),
unsigned: None,
// The following keys are dropped on conversion
room_id: Some(room_id.to_owned()),
prev_events: vec![],
depth: uint!(1),
auth_events: vec![],
redacts: None,
hashes: EventHash { sha256: String::new() },
signatures: None,
}
}
+494
View File
@@ -0,0 +1,494 @@
mod joined;
mod left;
mod state;
use std::{
cmp::{self},
collections::{BTreeMap, HashMap, HashSet},
time::Duration,
};
use axum::extract::State;
use conduwuit::{
Result, extract_variant,
utils::{
ReadyExt, TryFutureExtExt,
stream::{BroadbandExt, Tools, WidebandExt},
},
warn,
};
use conduwuit_service::Services;
use futures::{
FutureExt, StreamExt, TryFutureExt,
future::{OptionFuture, join3, join4, join5},
};
use ruma::{
DeviceId, OwnedUserId, RoomId, UserId,
api::client::{
filter::FilterDefinition,
sync::sync_events::{
self, DeviceLists,
v3::{
Filter, GlobalAccountData, InviteState, InvitedRoom, KnockState, KnockedRoom,
Presence, Rooms, ToDevice,
},
},
uiaa::UiaaResponse,
},
events::{
AnyRawAccountDataEvent,
presence::{PresenceEvent, PresenceEventContent},
},
serde::Raw,
};
use service::rooms::lazy_loading::{self, MemberSet, Options as _};
use super::{load_timeline, share_encrypted_room};
use crate::{
Ruma, RumaResponse,
client::{
is_ignored_invite,
sync::v3::{joined::load_joined_room, left::load_left_room},
},
};
/// The default maximum number of events to return in the `timeline` key of
/// joined and left rooms. If the number of events sent since the last sync
/// exceeds this number, the `timeline` will be `limited`.
const DEFAULT_TIMELINE_LIMIT: usize = 30;
/// A collection of updates to users' device lists, used for E2EE.
struct DeviceListUpdates {
changed: HashSet<OwnedUserId>,
left: HashSet<OwnedUserId>,
}
impl DeviceListUpdates {
fn new() -> Self {
Self {
changed: HashSet::new(),
left: HashSet::new(),
}
}
fn merge(&mut self, other: Self) {
self.changed.extend(other.changed);
self.left.extend(other.left);
}
fn is_empty(&self) -> bool { self.changed.is_empty() && self.left.is_empty() }
}
impl From<DeviceListUpdates> for DeviceLists {
fn from(val: DeviceListUpdates) -> Self {
Self {
changed: val.changed.into_iter().collect(),
left: val.left.into_iter().collect(),
}
}
}
/// References to common data needed to calculate the sync response.
#[derive(Clone, Copy)]
struct SyncContext<'a> {
/// The ID of the user requesting this sync.
syncing_user: &'a UserId,
/// The ID of the device requesting this sync, which will belong to
/// `syncing_user`.
syncing_device: &'a DeviceId,
/// The global count at the end of the previous sync response.
/// The previous sync's `current_count` will become the next sync's
/// `last_sync_end_count`. This will be None if no `since` query parameter
/// was specified, indicating an initial sync.
last_sync_end_count: Option<u64>,
/// The global count as of when we started building the sync response.
/// This is used as an upper bound when querying the database to ensure the
/// response represents a snapshot in time and doesn't include data which
/// appeared while the response was being built.
current_count: u64,
/// The `full_state` query parameter, used when syncing state for joined and
/// left rooms.
full_state: bool,
/// The sync filter, which the client uses to specify what data should be
/// included in the sync response.
filter: &'a FilterDefinition,
}
impl<'a> SyncContext<'a> {
fn lazy_loading_context(&self, room_id: &'a RoomId) -> lazy_loading::Context<'a> {
lazy_loading::Context {
user_id: self.syncing_user,
device_id: Some(self.syncing_device),
room_id,
token: self.last_sync_end_count,
options: Some(&self.filter.room.state.lazy_load_options),
}
}
#[inline]
fn lazy_loading_enabled(&self) -> bool {
(self.filter.room.state.lazy_load_options.is_enabled()
|| self.filter.room.timeline.lazy_load_options.is_enabled())
&& !self.full_state
}
}
type PresenceUpdates = HashMap<OwnedUserId, PresenceEventContent>;
/// # `GET /_matrix/client/r0/sync`
///
/// Synchronize the client's state with the latest state on the server.
///
/// - This endpoint takes a `since` parameter which should be the `next_batch`
/// value from a previous request for incremental syncs.
///
/// Calling this endpoint without a `since` parameter returns:
/// - Some of the most recent events of each timeline
/// - Notification counts for each room
/// - Joined and invited member counts, heroes
/// - All state events
///
/// Calling this endpoint with a `since` parameter from a previous `next_batch`
/// returns: For joined rooms:
/// - Some of the most recent events of each timeline that happened after since
/// - If user joined the room after since: All state events (unless lazy loading
/// is activated) and all device list updates in that room
/// - If the user was already in the room: A list of all events that are in the
/// state now, but were not in the state at `since`
/// - If the state we send contains a member event: Joined and invited member
/// counts, heroes
/// - Device list updates that happened after `since`
/// - If there are events in the timeline we send or the user send updated his
/// read mark: Notification counts
/// - EDUs that are active now (read receipts, typing updates, presence)
/// - TODO: Allow multiple sync streams to support Pantalaimon
///
/// For invited rooms:
/// - If the user was invited after `since`: A subset of the state of the room
/// at the point of the invite
///
/// For left rooms:
/// - If the user left after `since`: `prev_batch` token, empty state (TODO:
/// subset of the state at the point of the leave)
#[tracing::instrument(
name = "sync",
level = "debug",
skip_all,
fields(
since = %body.body.since.as_deref().unwrap_or_default(),
)
)]
pub(crate) async fn sync_events_route(
State(services): State<crate::State>,
body: Ruma<sync_events::v3::Request>,
) -> Result<sync_events::v3::Response, RumaResponse<UiaaResponse>> {
let (sender_user, sender_device) = body.sender();
// Presence update
if services.config.allow_local_presence {
services
.presence
.ping_presence(sender_user, &body.body.set_presence)
.await?;
}
// Setup watchers, so if there's no response, we can wait for them
let watcher = services.sync.watch(sender_user, sender_device);
let response = build_sync_events(&services, &body).await?;
if body.body.full_state
|| !(response.rooms.is_empty()
&& response.presence.is_empty()
&& response.account_data.is_empty()
&& response.device_lists.is_empty()
&& response.to_device.is_empty())
{
return Ok(response);
}
// Hang a few seconds so requests are not spammed
// Stop hanging if new info arrives
let default = Duration::from_secs(30);
let duration = cmp::min(body.body.timeout.unwrap_or(default), default);
_ = tokio::time::timeout(duration, watcher).await;
// Retry returning data
build_sync_events(&services, &body).await
}
pub(crate) async fn build_sync_events(
services: &Services,
body: &Ruma<sync_events::v3::Request>,
) -> Result<sync_events::v3::Response, RumaResponse<UiaaResponse>> {
let (syncing_user, syncing_device) = body.sender();
let current_count = services.globals.current_count()?;
// the `since` token is the last sync end count stringified
let last_sync_end_count = body
.body
.since
.as_ref()
.and_then(|string| string.parse().ok());
let full_state = body.body.full_state;
// FilterDefinition is very large (0x1000 bytes), let's put it on the heap
let filter = Box::new(match body.body.filter.as_ref() {
// use the default filter if none was specified
| None => FilterDefinition::default(),
// use inline filters directly
| Some(Filter::FilterDefinition(filter)) => filter.clone(),
// look up filter IDs from the database
| Some(Filter::FilterId(filter_id)) => services
.users
.get_filter(syncing_user, filter_id)
.await
.unwrap_or_default(),
});
let context = SyncContext {
syncing_user,
syncing_device,
last_sync_end_count,
current_count,
full_state,
filter: &filter,
};
let joined_rooms = services
.rooms
.state_cache
.rooms_joined(syncing_user)
.map(ToOwned::to_owned)
.broad_filter_map(|room_id| async {
let joined_room = load_joined_room(services, context, room_id.clone()).await;
match joined_room {
| Ok((room, updates)) => Some((room_id, room, updates)),
| Err(err) => {
warn!(?err, ?room_id, "error loading joined room {}", room_id);
None
},
}
})
.ready_fold(
(BTreeMap::new(), DeviceListUpdates::new()),
|(mut joined_rooms, mut all_updates), (room_id, joined_room, updates)| {
all_updates.merge(updates);
if !joined_room.is_empty() {
joined_rooms.insert(room_id, joined_room);
}
(joined_rooms, all_updates)
},
);
let left_rooms = services
.rooms
.state_cache
.rooms_left(syncing_user)
.broad_filter_map(|(room_id, leave_pdu)| {
load_left_room(services, context, room_id.clone(), leave_pdu)
.map_ok(move |left_room| (room_id, left_room))
.ok()
})
.ready_filter_map(|(room_id, left_room)| left_room.map(|left_room| (room_id, left_room)))
.collect();
let invited_rooms = services
.rooms
.state_cache
.rooms_invited(syncing_user)
.wide_filter_map(async |(room_id, invite_state)| {
if is_ignored_invite(services, syncing_user, &room_id).await {
None
} else {
Some((room_id, invite_state))
}
})
.fold_default(|mut invited_rooms: BTreeMap<_, _>, (room_id, invite_state)| async move {
let invite_count = services
.rooms
.state_cache
.get_invite_count(&room_id, syncing_user)
.await
.ok();
// only sync this invite if it was sent after the last /sync call
if last_sync_end_count < invite_count {
let invited_room = InvitedRoom {
invite_state: InviteState { events: invite_state },
};
invited_rooms.insert(room_id, invited_room);
}
invited_rooms
});
let knocked_rooms = services
.rooms
.state_cache
.rooms_knocked(syncing_user)
.fold_default(|mut knocked_rooms: BTreeMap<_, _>, (room_id, knock_state)| async move {
let knock_count = services
.rooms
.state_cache
.get_knock_count(&room_id, syncing_user)
.await
.ok();
// only sync this knock if it was sent after the last /sync call
if last_sync_end_count < knock_count {
let knocked_room = KnockedRoom {
knock_state: KnockState { events: knock_state },
};
knocked_rooms.insert(room_id, knocked_room);
}
knocked_rooms
});
let presence_updates: OptionFuture<_> = services
.config
.allow_local_presence
.then(|| process_presence_updates(services, last_sync_end_count, syncing_user))
.into();
let account_data = services
.account_data
.changes_since(None, syncing_user, last_sync_end_count, Some(current_count))
.ready_filter_map(|e| extract_variant!(e, AnyRawAccountDataEvent::Global))
.collect();
// Look for device list updates of this account
let keys_changed = services
.users
.keys_changed(syncing_user, last_sync_end_count, Some(current_count))
.map(ToOwned::to_owned)
.collect::<HashSet<_>>();
let to_device_events = services
.users
.get_to_device_events(
syncing_user,
syncing_device,
last_sync_end_count,
Some(current_count),
)
.collect::<Vec<_>>();
let device_one_time_keys_count = services
.users
.count_one_time_keys(syncing_user, syncing_device);
// Remove all to-device events the device received *last time*
let remove_to_device_events =
services
.users
.remove_to_device_events(syncing_user, syncing_device, last_sync_end_count);
let rooms = join4(joined_rooms, left_rooms, invited_rooms, knocked_rooms);
let ephemeral = join3(remove_to_device_events, to_device_events, presence_updates);
let top = join5(account_data, ephemeral, device_one_time_keys_count, keys_changed, rooms)
.boxed()
.await;
let (account_data, ephemeral, device_one_time_keys_count, keys_changed, rooms) = top;
let ((), to_device_events, presence_updates) = ephemeral;
let (joined_rooms, left_rooms, invited_rooms, knocked_rooms) = rooms;
let (joined_rooms, mut device_list_updates) = joined_rooms;
device_list_updates.changed.extend(keys_changed);
let response = sync_events::v3::Response {
account_data: GlobalAccountData { events: account_data },
device_lists: device_list_updates.into(),
device_one_time_keys_count,
// Fallback keys are not yet supported
device_unused_fallback_key_types: None,
next_batch: current_count.to_string(),
presence: Presence {
events: presence_updates
.into_iter()
.flat_map(IntoIterator::into_iter)
.map(|(sender, content)| PresenceEvent { content, sender })
.map(|ref event| Raw::new(event))
.filter_map(Result::ok)
.collect(),
},
rooms: Rooms {
leave: left_rooms,
join: joined_rooms,
invite: invited_rooms,
knock: knocked_rooms,
},
to_device: ToDevice { events: to_device_events },
};
Ok(response)
}
#[tracing::instrument(name = "presence", level = "debug", skip_all)]
async fn process_presence_updates(
services: &Services,
last_sync_end_count: Option<u64>,
syncing_user: &UserId,
) -> PresenceUpdates {
services
.presence
.presence_since(last_sync_end_count.unwrap_or(0)) // send all presences on initial sync
.filter(|(user_id, ..)| {
services
.rooms
.state_cache
.user_sees_user(syncing_user, user_id)
})
.filter_map(|(user_id, _, presence_bytes)| {
services
.presence
.from_json_bytes_to_event(presence_bytes, user_id)
.map_ok(move |event| (user_id, event))
.ok()
})
.map(|(user_id, event)| (user_id.to_owned(), event.content))
.collect()
.await
}
/// Using the provided sync context and an iterator of user IDs in the
/// `timeline`, return a HashSet of user IDs whose membership events should be
/// sent to the client if lazy-loading is enabled.
#[allow(clippy::let_and_return)]
async fn prepare_lazily_loaded_members(
services: &Services,
sync_context: SyncContext<'_>,
room_id: &RoomId,
timeline_members: impl Iterator<Item = OwnedUserId>,
) -> Option<MemberSet> {
let lazy_loading_context = &sync_context.lazy_loading_context(room_id);
// reset lazy loading state on initial sync.
// do this even if lazy loading is disabled so future lazy loads
// will have the correct members.
if sync_context.last_sync_end_count.is_none() {
services
.rooms
.lazy_loading
.reset(lazy_loading_context)
.await;
}
// filter the input members through `retain_lazy_members`, which
// contains the actual lazy loading logic.
let lazily_loaded_members =
OptionFuture::from(sync_context.lazy_loading_enabled().then(|| {
services
.rooms
.lazy_loading
.retain_lazy_members(timeline_members.collect(), lazy_loading_context)
}))
.await;
lazily_loaded_members
}
+280
View File
@@ -0,0 +1,280 @@
use std::{collections::BTreeSet, ops::ControlFlow};
use conduwuit::{
Result, at, is_equal_to,
matrix::{
Event,
pdu::{PduCount, PduEvent},
},
utils::{
BoolExt, IterStream, ReadyExt, TryFutureExtExt,
stream::{BroadbandExt, TryIgnore},
},
};
use conduwuit_service::{
Services,
rooms::{lazy_loading::MemberSet, short::ShortStateHash},
};
use futures::{FutureExt, StreamExt};
use itertools::Itertools;
use ruma::{OwnedEventId, RoomId, UserId, events::StateEventType};
use service::rooms::short::ShortEventId;
use tracing::trace;
use crate::client::TimelinePdus;
/// Calculate the state events to include in an initial sync response.
///
/// If lazy-loading is enabled (`lazily_loaded_members` is Some), the returned
/// Vec will include the membership events of exclusively the members in
/// `lazily_loaded_members`.
#[tracing::instrument(
name = "initial",
level = "trace",
skip_all,
fields(current_shortstatehash)
)]
#[allow(clippy::too_many_arguments)]
pub(super) async fn build_state_initial(
services: &Services,
sender_user: &UserId,
timeline_start_shortstatehash: ShortStateHash,
lazily_loaded_members: Option<&MemberSet>,
) -> Result<Vec<PduEvent>> {
// load the keys and event IDs of the state events at the start of the timeline
let (shortstatekeys, event_ids): (Vec<_>, Vec<_>) = services
.rooms
.state_accessor
.state_full_ids(timeline_start_shortstatehash)
.unzip()
.await;
trace!("performing initial sync of {} state events", event_ids.len());
services
.rooms
.short
// look up the full state keys
.multi_get_statekey_from_short(shortstatekeys.into_iter().stream())
.zip(event_ids.into_iter().stream())
.ready_filter_map(|item| Some((item.0.ok()?, item.1)))
.ready_filter_map(|((event_type, state_key), event_id)| {
if let Some(lazily_loaded_members) = lazily_loaded_members {
/*
if lazy loading is enabled, filter out membership events which aren't for a user
included in `lazily_loaded_members` or for the user requesting the sync.
*/
let event_is_redundant = event_type == StateEventType::RoomMember
&& state_key.as_str().try_into().is_ok_and(|user_id: &UserId| {
sender_user != user_id && !lazily_loaded_members.contains(user_id)
});
event_is_redundant.or_some(event_id)
} else {
Some(event_id)
}
})
.broad_filter_map(|event_id: OwnedEventId| async move {
services.rooms.timeline.get_pdu(&event_id).await.ok()
})
.collect()
.map(Ok)
.await
}
/// Calculate the state events to include in an incremental sync response.
///
/// If lazy-loading is enabled (`lazily_loaded_members` is Some), the returned
/// Vec will include the membership events of all the members in
/// `lazily_loaded_members`.
#[tracing::instrument(name = "incremental", level = "trace", skip_all)]
#[allow(clippy::too_many_arguments)]
pub(super) async fn build_state_incremental<'a>(
services: &Services,
sender_user: &'a UserId,
room_id: &RoomId,
last_sync_end_count: PduCount,
last_sync_end_shortstatehash: ShortStateHash,
timeline_start_shortstatehash: ShortStateHash,
timeline_end_shortstatehash: ShortStateHash,
timeline: &TimelinePdus,
lazily_loaded_members: Option<&'a MemberSet>,
) -> Result<Vec<PduEvent>> {
/*
NB: a limited sync is one where `timeline.limited == true`. Synapse calls this a "gappy" sync internally.
The algorithm implemented in this function is, currently, quite different from the algorithm vaguely described
by the Matrix specification. This is because the specification's description of the `state` property does not accurately
reflect how Synapse behaves, and therefore how client SDKs behave. Notable differences include:
1. We do not compute the delta using the naive approach of "every state event from the end of the last sync
up to the start of this sync's timeline". see below for details.
2. If lazy-loading is enabled, we include lazily-loaded membership events. The specific users to include are determined
elsewhere and supplied to this function in the `lazily_loaded_members` parameter.
*/
/*
the `state` property of an incremental sync which isn't limited are _usually_ empty.
(note: the specification says that the `state` property is _always_ empty for limited syncs, which is incorrect.)
however, if an event in the timeline (`timeline.pdus`) merges a split in the room's DAG (i.e. has multiple `prev_events`),
the state at the _end_ of the timeline may include state events which were merged in and don't exist in the state
at the _start_ of the timeline. because this is uncommon, we check here to see if any events in the timeline
merged a split in the DAG.
see: https://github.com/element-hq/synapse/issues/16941
*/
let timeline_is_linear = timeline.pdus.is_empty() || {
let last_pdu_of_last_sync = services
.rooms
.timeline
.pdus_rev(Some(sender_user), room_id, Some(last_sync_end_count.saturating_add(1)))
.boxed()
.next()
.await
.transpose()
.expect("last sync should have had some PDUs")
.map(at!(1));
// make sure the prev_events of each pdu in the timeline refer only to the
// previous pdu
timeline
.pdus
.iter()
.try_fold(last_pdu_of_last_sync.map(|pdu| pdu.event_id), |prev_event_id, (_, pdu)| {
if let Ok(pdu_prev_event_id) = pdu.prev_events.iter().exactly_one() {
if prev_event_id
.as_ref()
.is_none_or(is_equal_to!(pdu_prev_event_id))
{
return ControlFlow::Continue(Some(pdu_prev_event_id.to_owned()));
}
}
trace!(
"pdu {:?} has split prev_events (expected {:?}): {:?}",
pdu.event_id, prev_event_id, pdu.prev_events
);
ControlFlow::Break(())
})
.is_continue()
};
if timeline_is_linear && !timeline.limited {
// if there are no splits in the DAG and the timeline isn't limited, then
// `state` will always be empty unless lazy loading is enabled.
if let Some(lazily_loaded_members) = lazily_loaded_members {
if !timeline.pdus.is_empty() {
// lazy loading is enabled, so we return the membership events which were
// requested by the caller.
let lazy_membership_events: Vec<_> = lazily_loaded_members
.iter()
.stream()
.broad_filter_map(|user_id| async move {
if user_id == sender_user {
return None;
}
services
.rooms
.state_accessor
.state_get(
timeline_start_shortstatehash,
&StateEventType::RoomMember,
user_id.as_str(),
)
.ok()
.await
})
.collect()
.await;
if !lazy_membership_events.is_empty() {
trace!(
"syncing lazy membership events for members: {:?}",
lazy_membership_events
.iter()
.map(|pdu| pdu.state_key().unwrap())
.collect::<Vec<_>>()
);
}
return Ok(lazy_membership_events);
}
}
// lazy loading is disabled, `state` is empty.
return Ok(vec![]);
}
/*
at this point, either the timeline is `limited` or the DAG has a split in it. this necessitates
computing the incremental state (which may be empty).
NOTE: this code path does not use the `lazy_membership_events` parameter. any changes to membership will be included
in the incremental state. therefore, the incremental state may include "redundant" membership events,
which we do not filter out because A. the spec forbids lazy-load filtering if the timeline is `limited`,
and B. DAG splits which require sending extra membership state events are (probably) uncommon enough that
the performance penalty is acceptable.
*/
trace!(?timeline_is_linear, ?timeline.limited, "computing state for incremental sync");
// fetch the shorteventids of state events in the timeline
let state_events_in_timeline: BTreeSet<ShortEventId> = services
.rooms
.short
.multi_get_or_create_shorteventid(timeline.pdus.iter().filter_map(|(_, pdu)| {
if pdu.state_key().is_some() {
Some(pdu.event_id.as_ref())
} else {
None
}
}))
.collect()
.await;
trace!("{} state events in timeline", state_events_in_timeline.len());
/*
fetch the state events which were added since the last sync.
specifically we fetch the difference between the state at the last sync and the state at the _end_
of the timeline, and then we filter out state events in the timeline itself using the shorteventids we fetched.
this is necessary to account for splits in the DAG, as explained above.
*/
let state_diff = services
.rooms
.short
.multi_get_eventid_from_short::<'_, OwnedEventId, _>(
services
.rooms
.state_accessor
.state_added((last_sync_end_shortstatehash, timeline_end_shortstatehash))
.await?
.stream()
.ready_filter_map(|(_, shorteventid)| {
if state_events_in_timeline.contains(&shorteventid) {
None
} else {
Some(shorteventid)
}
}),
)
.ignore_err();
// finally, fetch the PDU contents and collect them into a vec
let state_diff_pdus = state_diff
.broad_filter_map(|event_id| async move {
services
.rooms
.timeline
.get_non_outlier_pdu(&event_id)
.await
.ok()
})
.collect::<Vec<_>>()
.await;
trace!(?state_diff_pdus, "collected state PDUs for incremental sync");
Ok(state_diff_pdus)
}
-848
View File
@@ -1,848 +0,0 @@
use std::{
cmp::{self, Ordering},
collections::{BTreeMap, BTreeSet, HashMap, HashSet},
time::Duration,
};
use axum::extract::State;
use conduwuit::{
Err, Error, Event, PduCount, Result, at, debug, error, extract_variant,
matrix::TypeStateKey,
utils::{
BoolExt, IterStream, ReadyExt, TryFutureExtExt,
math::{ruma_from_usize, usize_from_ruma, usize_from_u64_truncated},
stream::WidebandExt,
},
warn,
};
use conduwuit_service::{
Services,
rooms::read_receipt::pack_receipts,
sync::{into_db_key, into_snake_key},
};
use futures::{FutureExt, StreamExt, TryFutureExt};
use ruma::{
MilliSecondsSinceUnixEpoch, OwnedEventId, OwnedRoomId, RoomId, UInt, UserId,
api::client::sync::sync_events::{
self, DeviceLists, UnreadNotificationsCount,
v4::{SlidingOp, SlidingSyncRoomHero},
},
directory::RoomTypeFilter,
events::{
AnyRawAccountDataEvent, AnySyncEphemeralRoomEvent, StateEventType,
TimelineEventType::*,
room::member::{MembershipState, RoomMemberEventContent},
},
serde::Raw,
uint,
};
use super::{load_timeline, share_encrypted_room};
use crate::{
Ruma,
client::{DEFAULT_BUMP_TYPES, ignored_filter, is_ignored_invite},
};
type TodoRooms = BTreeMap<OwnedRoomId, (BTreeSet<TypeStateKey>, usize, u64)>;
const SINGLE_CONNECTION_SYNC: &str = "single_connection_sync";
#[allow(clippy::cognitive_complexity)]
/// POST `/_matrix/client/unstable/org.matrix.msc3575/sync`
///
/// Sliding Sync endpoint (future endpoint: `/_matrix/client/v4/sync`)
pub(crate) async fn sync_events_v4_route(
State(services): State<crate::State>,
body: Ruma<sync_events::v4::Request>,
) -> Result<sync_events::v4::Response> {
debug_assert!(DEFAULT_BUMP_TYPES.is_sorted(), "DEFAULT_BUMP_TYPES is not sorted");
let sender_user = body.sender_user.as_ref().expect("user is authenticated");
let sender_device = body.sender_device.as_ref().expect("user is authenticated");
let mut body = body.body;
// Setup watchers, so if there's no response, we can wait for them
let watcher = services.sync.watch(sender_user, sender_device);
let next_batch = services.globals.next_count()?;
let conn_id = body
.conn_id
.clone()
.unwrap_or_else(|| SINGLE_CONNECTION_SYNC.to_owned());
let globalsince = body
.pos
.as_ref()
.and_then(|string| string.parse().ok())
.unwrap_or(0);
let db_key = into_db_key(sender_user, sender_device, conn_id.clone());
if globalsince != 0 && !services.sync.remembered(&db_key) {
debug!("Restarting sync stream because it was gone from the database");
return Err!(Request(UnknownPos("Connection data lost since last time")));
}
if globalsince == 0 {
services.sync.forget_sync_request_connection(&db_key);
}
// Get sticky parameters from cache
let snake_key = into_snake_key(sender_user, sender_device, conn_id.clone());
let known_rooms = services
.sync
.update_sync_request_with_cache(&snake_key, &mut body);
let all_joined_rooms: Vec<_> = services
.rooms
.state_cache
.rooms_joined(sender_user)
.map(ToOwned::to_owned)
.collect()
.await;
let all_invited_rooms: Vec<_> = services
.rooms
.state_cache
.rooms_invited(sender_user)
.wide_filter_map(async |(room_id, invite_state)| {
if is_ignored_invite(&services, sender_user, &room_id).await {
None
} else {
Some((room_id, invite_state))
}
})
.map(|r| r.0)
.collect()
.await;
let all_knocked_rooms: Vec<_> = services
.rooms
.state_cache
.rooms_knocked(sender_user)
.map(|r| r.0)
.collect()
.await;
let all_invited_rooms: Vec<&RoomId> = all_invited_rooms.iter().map(AsRef::as_ref).collect();
let all_knocked_rooms: Vec<&RoomId> = all_knocked_rooms.iter().map(AsRef::as_ref).collect();
let all_rooms: Vec<&RoomId> = all_joined_rooms
.iter()
.map(AsRef::as_ref)
.chain(all_invited_rooms.iter().map(AsRef::as_ref))
.chain(all_knocked_rooms.iter().map(AsRef::as_ref))
.collect();
let all_joined_rooms = all_joined_rooms.iter().map(AsRef::as_ref).collect();
let all_invited_rooms = all_invited_rooms.iter().map(AsRef::as_ref).collect();
if body.extensions.to_device.enabled.unwrap_or(false) {
services
.users
.remove_to_device_events(sender_user, sender_device, globalsince)
.await;
}
let mut left_encrypted_users = HashSet::new(); // Users that have left any encrypted rooms the sender was in
let mut device_list_changes = HashSet::new();
let mut device_list_left = HashSet::new();
let mut receipts = sync_events::v4::Receipts { rooms: BTreeMap::new() };
let mut account_data = sync_events::v4::AccountData {
global: Vec::new(),
rooms: BTreeMap::new(),
};
if body.extensions.account_data.enabled.unwrap_or(false) {
account_data.global = services
.account_data
.changes_since(None, sender_user, globalsince, Some(next_batch))
.ready_filter_map(|e| extract_variant!(e, AnyRawAccountDataEvent::Global))
.collect()
.await;
if let Some(rooms) = body.extensions.account_data.rooms {
for room in rooms {
account_data.rooms.insert(
room.clone(),
services
.account_data
.changes_since(Some(&room), sender_user, globalsince, Some(next_batch))
.ready_filter_map(|e| extract_variant!(e, AnyRawAccountDataEvent::Room))
.collect()
.await,
);
}
}
}
if body.extensions.e2ee.enabled.unwrap_or(false) {
// Look for device list updates of this account
device_list_changes.extend(
services
.users
.keys_changed(sender_user, globalsince, None)
.map(ToOwned::to_owned)
.collect::<Vec<_>>()
.await,
);
for room_id in &all_joined_rooms {
let room_id: &&RoomId = room_id;
let Ok(current_shortstatehash) =
services.rooms.state.get_room_shortstatehash(room_id).await
else {
error!("Room {room_id} has no state");
continue;
};
let since_shortstatehash = services
.rooms
.user
.get_token_shortstatehash(room_id, globalsince)
.await
.ok();
let encrypted_room = services
.rooms
.state_accessor
.state_get(current_shortstatehash, &StateEventType::RoomEncryption, "")
.await
.is_ok();
if let Some(since_shortstatehash) = since_shortstatehash {
// Skip if there are only timeline changes
if since_shortstatehash == current_shortstatehash {
continue;
}
let since_encryption = services
.rooms
.state_accessor
.state_get(since_shortstatehash, &StateEventType::RoomEncryption, "")
.await;
let since_sender_member: Option<RoomMemberEventContent> = services
.rooms
.state_accessor
.state_get_content(
since_shortstatehash,
&StateEventType::RoomMember,
sender_user.as_str(),
)
.ok()
.await;
let joined_since_last_sync = since_sender_member
.as_ref()
.is_none_or(|member| member.membership != MembershipState::Join);
let new_encrypted_room = encrypted_room && since_encryption.is_err();
if encrypted_room {
let current_state_ids: HashMap<_, OwnedEventId> = services
.rooms
.state_accessor
.state_full_ids(current_shortstatehash)
.collect()
.await;
let since_state_ids: HashMap<_, _> = services
.rooms
.state_accessor
.state_full_ids(since_shortstatehash)
.collect()
.await;
for (key, id) in current_state_ids {
if since_state_ids.get(&key) != Some(&id) {
let Ok(pdu) = services.rooms.timeline.get_pdu(&id).await else {
error!("Pdu in state not found: {id}");
continue;
};
if pdu.kind == RoomMember {
if let Some(Ok(user_id)) =
pdu.state_key.as_deref().map(UserId::parse)
{
if user_id == sender_user {
continue;
}
let content: RoomMemberEventContent = pdu.get_content()?;
match content.membership {
| MembershipState::Join => {
// A new user joined an encrypted room
if !share_encrypted_room(
&services,
sender_user,
user_id,
Some(room_id),
)
.await
{
device_list_changes.insert(user_id.to_owned());
}
},
| MembershipState::Leave => {
// Write down users that have left encrypted rooms we
// are in
left_encrypted_users.insert(user_id.to_owned());
},
| _ => {},
}
}
}
}
}
if joined_since_last_sync || new_encrypted_room {
// If the user is in a new encrypted room, give them all joined users
device_list_changes.extend(
services
.rooms
.state_cache
.room_members(room_id)
// Don't send key updates from the sender to the sender
.ready_filter(|&user_id| sender_user != user_id)
// Only send keys if the sender doesn't share an encrypted room with the target
// already
.filter_map(|user_id| {
share_encrypted_room(&services, sender_user, user_id, Some(room_id))
.map(|res| res.or_some(user_id.to_owned()))
})
.collect::<Vec<_>>()
.await,
);
}
}
}
// Look for device list updates in this room
device_list_changes.extend(
services
.users
.room_keys_changed(room_id, globalsince, None)
.map(|(user_id, _)| user_id)
.map(ToOwned::to_owned)
.collect::<Vec<_>>()
.await,
);
}
for user_id in left_encrypted_users {
let dont_share_encrypted_room =
!share_encrypted_room(&services, sender_user, &user_id, None).await;
// If the user doesn't share an encrypted room with the target anymore, we need
// to tell them
if dont_share_encrypted_room {
device_list_left.insert(user_id);
}
}
}
let mut lists = BTreeMap::new();
let mut todo_rooms: TodoRooms = BTreeMap::new(); // and required state
for (list_id, list) in &body.lists {
let active_rooms = match list.filters.clone().and_then(|f| f.is_invite) {
| Some(true) => &all_invited_rooms,
| Some(false) => &all_joined_rooms,
| None => &all_rooms,
};
let active_rooms = match list.filters.clone().map(|f| f.not_room_types) {
| Some(filter) if filter.is_empty() => active_rooms.clone(),
| Some(value) => filter_rooms(&services, active_rooms, &value, true).await,
| None => active_rooms.clone(),
};
let active_rooms = match list.filters.clone().map(|f| f.room_types) {
| Some(filter) if filter.is_empty() => active_rooms.clone(),
| Some(value) => filter_rooms(&services, &active_rooms, &value, false).await,
| None => active_rooms,
};
let mut new_known_rooms: BTreeSet<OwnedRoomId> = BTreeSet::new();
let ranges = list.ranges.clone();
lists.insert(list_id.clone(), sync_events::v4::SyncList {
ops: ranges
.into_iter()
.map(|mut r| {
r.0 = r.0.clamp(
uint!(0),
UInt::try_from(active_rooms.len().saturating_sub(1)).unwrap_or(UInt::MAX),
);
r.1 = r.1.clamp(
r.0,
UInt::try_from(active_rooms.len().saturating_sub(1)).unwrap_or(UInt::MAX),
);
let room_ids = if !active_rooms.is_empty() {
active_rooms[usize_from_ruma(r.0)..=usize_from_ruma(r.1)].to_vec()
} else {
Vec::new()
};
new_known_rooms.extend(room_ids.clone().into_iter().map(ToOwned::to_owned));
for room_id in &room_ids {
let todo_room = todo_rooms.entry((*room_id).to_owned()).or_insert((
BTreeSet::new(),
0_usize,
u64::MAX,
));
let limit: usize = list
.room_details
.timeline_limit
.map(u64::from)
.map_or(10, usize_from_u64_truncated)
.min(100);
todo_room.0.extend(
list.room_details
.required_state
.iter()
.map(|(ty, sk)| (ty.clone(), sk.as_str().into())),
);
todo_room.1 = todo_room.1.max(limit);
// 0 means unknown because it got out of date
todo_room.2 = todo_room.2.min(
known_rooms
.get(list_id.as_str())
.and_then(|k| k.get(*room_id))
.copied()
.unwrap_or(0),
);
}
sync_events::v4::SyncOp {
op: SlidingOp::Sync,
range: Some(r),
index: None,
room_ids: room_ids.into_iter().map(ToOwned::to_owned).collect(),
room_id: None,
}
})
.collect(),
count: ruma_from_usize(active_rooms.len()),
});
if let Some(conn_id) = &body.conn_id {
let db_key = into_db_key(sender_user, sender_device, conn_id);
services.sync.update_sync_known_rooms(
&db_key,
list_id.clone(),
new_known_rooms,
globalsince,
);
}
}
let mut known_subscription_rooms = BTreeSet::new();
for (room_id, room) in &body.room_subscriptions {
if !services.rooms.metadata.exists(room_id).await
|| services.rooms.metadata.is_disabled(room_id).await
|| services.rooms.metadata.is_banned(room_id).await
{
continue;
}
let todo_room =
todo_rooms
.entry(room_id.clone())
.or_insert((BTreeSet::new(), 0_usize, u64::MAX));
let limit: usize = room
.timeline_limit
.map(u64::from)
.map_or(10, usize_from_u64_truncated)
.min(100);
todo_room.0.extend(
room.required_state
.iter()
.map(|(ty, sk)| (ty.clone(), sk.as_str().into())),
);
todo_room.1 = todo_room.1.max(limit);
// 0 means unknown because it got out of date
todo_room.2 = todo_room.2.min(
known_rooms
.get("subscriptions")
.and_then(|k| k.get(room_id))
.copied()
.unwrap_or(0),
);
known_subscription_rooms.insert(room_id.clone());
}
for r in body.unsubscribe_rooms {
known_subscription_rooms.remove(&r);
body.room_subscriptions.remove(&r);
}
if let Some(conn_id) = &body.conn_id {
let db_key = into_db_key(sender_user, sender_device, conn_id);
services.sync.update_sync_known_rooms(
&db_key,
"subscriptions".to_owned(),
known_subscription_rooms,
globalsince,
);
}
if let Some(conn_id) = body.conn_id.clone() {
let db_key = into_db_key(sender_user, sender_device, conn_id);
services
.sync
.update_sync_subscriptions(&db_key, body.room_subscriptions);
}
let mut rooms = BTreeMap::new();
for (room_id, (required_state_request, timeline_limit, roomsince)) in &todo_rooms {
let roomsincecount = PduCount::Normal(*roomsince);
let mut timestamp: Option<_> = None;
let mut invite_state = None;
let (timeline_pdus, limited);
let new_room_id: &RoomId = (*room_id).as_ref();
if all_invited_rooms.contains(&new_room_id) {
// TODO: figure out a timestamp we can use for remote invites
invite_state = services
.rooms
.state_cache
.invite_state(sender_user, room_id)
.await
.ok();
(timeline_pdus, limited) = (Vec::new(), true);
} else {
(timeline_pdus, limited) = match load_timeline(
&services,
sender_user,
room_id,
roomsincecount,
None,
*timeline_limit,
)
.await
{
| Ok(value) => value,
| Err(err) => {
warn!("Encountered missing timeline in {}, error {}", room_id, err);
continue;
},
};
}
account_data.rooms.insert(
room_id.to_owned(),
services
.account_data
.changes_since(Some(room_id), sender_user, *roomsince, Some(next_batch))
.ready_filter_map(|e| extract_variant!(e, AnyRawAccountDataEvent::Room))
.collect()
.await,
);
let last_privateread_update = services
.rooms
.read_receipt
.last_privateread_update(sender_user, room_id)
.await > *roomsince;
let private_read_event = if last_privateread_update {
services
.rooms
.read_receipt
.private_read_get(room_id, sender_user)
.await
.ok()
} else {
None
};
let mut vector: Vec<Raw<AnySyncEphemeralRoomEvent>> = services
.rooms
.read_receipt
.readreceipts_since(room_id, *roomsince)
.filter_map(|(read_user, _ts, v)| async move {
services
.users
.user_is_ignored(read_user, sender_user)
.await
.or_some(v)
})
.collect()
.await;
if let Some(private_read_event) = private_read_event {
vector.push(private_read_event);
}
let receipt_size = vector.len();
receipts
.rooms
.insert(room_id.clone(), pack_receipts(Box::new(vector.into_iter())));
if roomsince != &0
&& timeline_pdus.is_empty()
&& account_data.rooms.get(room_id).is_some_and(Vec::is_empty)
&& receipt_size == 0
{
continue;
}
let prev_batch = timeline_pdus
.first()
.map_or(Ok::<_, Error>(None), |(pdu_count, _)| {
Ok(Some(match pdu_count {
| PduCount::Backfilled(_) => {
error!("timeline in backfill state?!");
"0".to_owned()
},
| PduCount::Normal(c) => c.to_string(),
}))
})?
.or_else(|| {
if roomsince != &0 {
Some(roomsince.to_string())
} else {
None
}
});
let room_events: Vec<_> = timeline_pdus
.iter()
.stream()
.filter_map(|item| ignored_filter(&services, item.clone(), sender_user))
.map(at!(1))
.map(Event::into_format)
.collect()
.await;
for (_, pdu) in timeline_pdus {
let ts = MilliSecondsSinceUnixEpoch(pdu.origin_server_ts);
if DEFAULT_BUMP_TYPES.binary_search(&pdu.kind).is_ok()
&& timestamp.is_none_or(|time| time <= ts)
{
timestamp = Some(ts);
}
}
let required_state = required_state_request
.iter()
.stream()
.filter_map(|state| async move {
services
.rooms
.state_accessor
.room_state_get(room_id, &state.0, &state.1)
.await
.map(Event::into_format)
.ok()
})
.collect()
.await;
// Heroes
let heroes: Vec<_> = services
.rooms
.state_cache
.room_members(room_id)
.ready_filter(|&member| member != sender_user)
.filter_map(|user_id| {
services
.rooms
.state_accessor
.get_member(room_id, user_id)
.map_ok(|memberevent| SlidingSyncRoomHero {
user_id: user_id.into(),
name: memberevent.displayname,
avatar: memberevent.avatar_url,
})
.ok()
})
.take(5)
.collect()
.await;
let name = match heroes.len().cmp(&(1_usize)) {
| Ordering::Greater => {
let firsts = heroes[1..]
.iter()
.map(|h| h.name.clone().unwrap_or_else(|| h.user_id.to_string()))
.collect::<Vec<_>>()
.join(", ");
let last = heroes[0]
.name
.clone()
.unwrap_or_else(|| heroes[0].user_id.to_string());
Some(format!("{firsts} and {last}"))
},
| Ordering::Equal => Some(
heroes[0]
.name
.clone()
.unwrap_or_else(|| heroes[0].user_id.to_string()),
),
| Ordering::Less => None,
};
let heroes_avatar = if heroes.len() == 1 {
heroes[0].avatar.clone()
} else {
None
};
rooms.insert(room_id.clone(), sync_events::v4::SlidingSyncRoom {
name: services
.rooms
.state_accessor
.get_name(room_id)
.await
.ok()
.or(name),
avatar: match heroes_avatar {
| Some(heroes_avatar) => ruma::JsOption::Some(heroes_avatar),
| _ => match services.rooms.state_accessor.get_avatar(room_id).await {
| ruma::JsOption::Some(avatar) => ruma::JsOption::from_option(avatar.url),
| ruma::JsOption::Null => ruma::JsOption::Null,
| ruma::JsOption::Undefined => ruma::JsOption::Undefined,
},
},
initial: Some(roomsince == &0),
is_dm: None,
invite_state,
unread_notifications: UnreadNotificationsCount {
highlight_count: Some(
services
.rooms
.user
.highlight_count(sender_user, room_id)
.await
.try_into()
.expect("notification count can't go that high"),
),
notification_count: Some(
services
.rooms
.user
.notification_count(sender_user, room_id)
.await
.try_into()
.expect("notification count can't go that high"),
),
},
timeline: room_events,
required_state,
prev_batch,
limited,
joined_count: Some(
services
.rooms
.state_cache
.room_joined_count(room_id)
.await
.unwrap_or(0)
.try_into()
.unwrap_or_else(|_| uint!(0)),
),
invited_count: Some(
services
.rooms
.state_cache
.room_invited_count(room_id)
.await
.unwrap_or(0)
.try_into()
.unwrap_or_else(|_| uint!(0)),
),
num_live: None, // Count events in timeline greater than global sync counter
timestamp,
heroes: Some(heroes),
});
}
if rooms.iter().all(|(id, r)| {
r.timeline.is_empty() && r.required_state.is_empty() && !receipts.rooms.contains_key(id)
}) {
// Hang a few seconds so requests are not spammed
// Stop hanging if new info arrives
let default = Duration::from_secs(30);
let duration = cmp::min(body.timeout.unwrap_or(default), default);
_ = tokio::time::timeout(duration, watcher).await;
}
Ok(sync_events::v4::Response {
initial: globalsince == 0,
txn_id: body.txn_id.clone(),
pos: next_batch.to_string(),
lists,
rooms,
extensions: sync_events::v4::Extensions {
to_device: if body.extensions.to_device.enabled.unwrap_or(false) {
Some(sync_events::v4::ToDevice {
events: services
.users
.get_to_device_events(
sender_user,
sender_device,
Some(globalsince),
Some(next_batch),
)
.collect()
.await,
next_batch: next_batch.to_string(),
})
} else {
None
},
e2ee: sync_events::v4::E2EE {
device_lists: DeviceLists {
changed: device_list_changes.into_iter().collect(),
left: device_list_left.into_iter().collect(),
},
device_one_time_keys_count: services
.users
.count_one_time_keys(sender_user, sender_device)
.await,
// Fallback keys are not yet supported
device_unused_fallback_key_types: None,
},
account_data,
receipts,
typing: sync_events::v4::Typing { rooms: BTreeMap::new() },
},
delta_token: None,
})
}
async fn filter_rooms<'a>(
services: &Services,
rooms: &[&'a RoomId],
filter: &[RoomTypeFilter],
negate: bool,
) -> Vec<&'a RoomId> {
rooms
.iter()
.stream()
.filter_map(|r| async move {
let room_type = services.rooms.state_accessor.get_room_type(r).await;
if room_type.as_ref().is_err_and(|e| !e.is_not_found()) {
return None;
}
let room_type_filter = RoomTypeFilter::from(room_type.ok());
let include = if negate {
!filter.contains(&room_type_filter)
} else {
filter.is_empty() || filter.contains(&room_type_filter)
};
include.then_some(r)
})
.collect()
.await
}
+78 -13
View File
@@ -1,6 +1,6 @@
use std::{
cmp::{self, Ordering},
collections::{BTreeMap, BTreeSet, HashMap, HashSet},
collections::{BTreeMap, BTreeSet, HashMap, HashSet, VecDeque},
ops::Deref,
time::Duration,
};
@@ -31,6 +31,7 @@
events::{
AnyRawAccountDataEvent, AnySyncEphemeralRoomEvent, StateEventType, TimelineEventType,
room::member::{MembershipState, RoomMemberEventContent},
typing::TypingEventContent,
},
serde::Raw,
uint,
@@ -39,7 +40,9 @@
use super::share_encrypted_room;
use crate::{
Ruma,
client::{DEFAULT_BUMP_TYPES, ignored_filter, is_ignored_invite, sync::load_timeline},
client::{
DEFAULT_BUMP_TYPES, TimelinePdus, ignored_filter, is_ignored_invite, sync::load_timeline,
},
};
type SyncInfo<'a> = (&'a UserId, &'a DeviceId, u64, &'a sync_events::v5::Request);
@@ -210,6 +213,9 @@ pub(crate) async fn sync_events_v5_route(
_ = tokio::time::timeout(duration, watcher).await;
}
let typing = collect_typing_events(services, sender_user, &body, &todo_rooms).await?;
response.extensions.typing = typing;
trace!(
rooms = ?response.rooms.len(),
account_data = ?response.extensions.account_data.rooms.len(),
@@ -293,6 +299,8 @@ async fn handle_lists<'a, Rooms, AllRooms>(
Rooms: Iterator<Item = &'a RoomId> + Clone + Send + 'a,
AllRooms: Iterator<Item = &'a RoomId> + Clone + Send + 'a,
{
// TODO MSC4186: Implement remaining list filters: is_dm, is_encrypted,
// room_types.
for (list_id, list) in &body.lists {
let active_rooms: Vec<_> = match list.filters.as_ref().and_then(|f| f.is_invite) {
| None => all_rooms.clone().collect(),
@@ -320,6 +328,7 @@ async fn handle_lists<'a, Rooms, AllRooms>(
for mut range in ranges {
range.0 = uint!(0);
range.1 = range.1.checked_add(uint!(1)).unwrap_or(range.1);
range.1 = range
.1
.clamp(range.0, UInt::try_from(active_rooms.len()).unwrap_or(UInt::MAX));
@@ -408,13 +417,13 @@ async fn process_rooms<'a, Rooms>(
.await
.ok();
(timeline_pdus, limited) = (Vec::new(), true);
(timeline_pdus, limited) = (VecDeque::new(), true);
} else {
(timeline_pdus, limited) = match load_timeline(
TimelinePdus { pdus: timeline_pdus, limited } = match load_timeline(
services,
sender_user,
room_id,
roomsincecount,
Some(roomsincecount),
Some(PduCount::from(next_batch)),
*timeline_limit,
)
@@ -433,7 +442,7 @@ async fn process_rooms<'a, Rooms>(
room_id.to_owned(),
services
.account_data
.changes_since(Some(room_id), sender_user, *roomsince, Some(next_batch))
.changes_since(Some(room_id), sender_user, Some(*roomsince), Some(next_batch))
.ready_filter_map(|e| extract_variant!(e, AnyRawAccountDataEvent::Room))
.collect()
.await,
@@ -459,11 +468,11 @@ async fn process_rooms<'a, Rooms>(
let mut receipts: Vec<Raw<AnySyncEphemeralRoomEvent>> = services
.rooms
.read_receipt
.readreceipts_since(room_id, *roomsince)
.readreceipts_since(room_id, Some(*roomsince))
.filter_map(|(read_user, _ts, v)| async move {
services
.users
.user_is_ignored(read_user, sender_user)
.user_is_ignored(&read_user, sender_user)
.await
.or_some(v)
})
@@ -498,7 +507,7 @@ async fn process_rooms<'a, Rooms>(
}
let prev_batch = timeline_pdus
.first()
.front()
.map_or(Ok::<_, Error>(None), |(pdu_count, _)| {
Ok(Some(match pdu_count {
| PduCount::Backfilled(_) => {
@@ -671,6 +680,62 @@ async fn process_rooms<'a, Rooms>(
}
Ok(rooms)
}
async fn collect_typing_events(
services: &Services,
sender_user: &UserId,
body: &sync_events::v5::Request,
todo_rooms: &TodoRooms,
) -> Result<sync_events::v5::response::Typing> {
if !body.extensions.typing.enabled.unwrap_or(false) {
return Ok(sync_events::v5::response::Typing::default());
}
let rooms: Vec<_> = body.extensions.typing.rooms.clone().unwrap_or_else(|| {
body.room_subscriptions
.keys()
.map(ToOwned::to_owned)
.collect()
});
let lists: Vec<_> = body
.extensions
.typing
.lists
.clone()
.unwrap_or_else(|| body.lists.keys().map(ToOwned::to_owned).collect::<Vec<_>>());
if rooms.is_empty() && lists.is_empty() {
return Ok(sync_events::v5::response::Typing::default());
}
let mut typing_response = sync_events::v5::response::Typing::default();
for (room_id, (_, _, roomsince)) in todo_rooms {
if services.rooms.typing.last_typing_update(room_id).await? <= *roomsince {
continue;
}
match services
.rooms
.typing
.typing_users_for_user(room_id, sender_user)
.await
{
| Ok(typing_users) => {
typing_response.rooms.insert(
room_id.to_owned(), // Already OwnedRoomId
Raw::new(&sync_events::v5::response::SyncTypingEvent {
content: TypingEventContent::new(typing_users),
})?,
);
},
| Err(e) => {
warn!(%room_id, "Failed to get typing events for room: {}", e);
},
}
}
Ok(typing_response)
}
async fn collect_account_data(
services: &Services,
(sender_user, _, globalsince, body): (&UserId, &DeviceId, u64, &sync_events::v5::Request),
@@ -686,7 +751,7 @@ async fn collect_account_data(
account_data.global = services
.account_data
.changes_since(None, sender_user, globalsince, None)
.changes_since(None, sender_user, Some(globalsince), None)
.ready_filter_map(|e| extract_variant!(e, AnyRawAccountDataEvent::Global))
.collect()
.await;
@@ -697,7 +762,7 @@ async fn collect_account_data(
room.clone(),
services
.account_data
.changes_since(Some(room), sender_user, globalsince, None)
.changes_since(Some(room), sender_user, Some(globalsince), None)
.ready_filter_map(|e| extract_variant!(e, AnyRawAccountDataEvent::Room))
.collect()
.await,
@@ -731,7 +796,7 @@ async fn collect_e2ee<'a, Rooms>(
device_list_changes.extend(
services
.users
.keys_changed(sender_user, globalsince, None)
.keys_changed(sender_user, Some(globalsince), None)
.map(ToOwned::to_owned)
.collect::<Vec<_>>()
.await,
@@ -867,7 +932,7 @@ async fn collect_e2ee<'a, Rooms>(
device_list_changes.extend(
services
.users
.room_keys_changed(room_id, globalsince, None)
.room_keys_changed(room_id, Some(globalsince), None)
.map(|(user_id, _)| user_id)
.map(ToOwned::to_owned)
.collect::<Vec<_>>()
-1
View File
@@ -52,7 +52,6 @@ pub(crate) async fn get_supported_versions_route(
("org.matrix.msc3026.busy_presence".to_owned(), true), /* busy presence status (https://github.com/matrix-org/matrix-spec-proposals/pull/3026) */
("org.matrix.msc3827".to_owned(), true), /* filtering of /publicRooms by room type (https://github.com/matrix-org/matrix-spec-proposals/pull/3827) */
("org.matrix.msc3952_intentional_mentions".to_owned(), true), /* intentional mentions (https://github.com/matrix-org/matrix-spec-proposals/pull/3952) */
("org.matrix.msc3575".to_owned(), true), /* sliding sync (https://github.com/matrix-org/matrix-spec-proposals/pull/3575/files#r1588877046) */
("org.matrix.msc3916.stable".to_owned(), true), /* authenticated media (https://github.com/matrix-org/matrix-spec-proposals/pull/3916) */
("org.matrix.msc4180".to_owned(), true), /* stable flag for 3916 (https://github.com/matrix-org/matrix-spec-proposals/pull/4180) */
("uk.tcpip.msc4133".to_owned(), true), /* Extending User Profile API with Key:Value Pairs (https://github.com/matrix-org/matrix-spec-proposals/pull/4133) */
+1 -1
View File
@@ -78,7 +78,7 @@ pub(crate) async fn well_known_support(
while let Some(user_id) = stream.next().await {
// Skip server user
if *user_id == services.globals.server_user {
break;
continue;
}
contacts.push(Contact {
role: role_value.clone(),
+1 -1
View File
@@ -143,7 +143,6 @@ pub fn build(router: Router<State>, server: &Server) -> Router<State> {
.put(client::send_state_event_for_empty_key_route),
)
.ruma_route(&client::sync_events_route)
.ruma_route(&client::sync_events_v4_route)
.ruma_route(&client::sync_events_v5_route)
.ruma_route(&client::get_context_route)
.ruma_route(&client::get_message_events_route)
@@ -226,6 +225,7 @@ pub fn build(router: Router<State>, server: &Server) -> Router<State> {
.ruma_route(&server::well_known_server)
.ruma_route(&server::get_content_route)
.ruma_route(&server::get_content_thumbnail_route)
.ruma_route(&server::get_edutypes_route)
.route("/_conduwuit/local_user_count", get(client::conduwuit_local_user_count))
.route("/_continuwuity/local_user_count", get(client::conduwuit_local_user_count));
} else {
+13
View File
@@ -34,6 +34,19 @@ pub(super) async fn from(
let max_body_size = services.server.config.max_request_size;
// Check if the Content-Length header is present and valid, saves us streaming
// the response into memory
if let Some(content_length) = parts.headers.get(http::header::CONTENT_LENGTH) {
if let Ok(content_length) = content_length
.to_str()
.map(|s| s.parse::<usize>().unwrap_or_default())
{
if content_length > max_body_size {
return Err(err!(Request(TooLarge("Request body too large"))));
}
}
}
let body = axum::body::to_bytes(body, max_body_size)
.await
.map_err(|e| err!(Request(TooLarge("Request body too large: {e}"))))?;
+19
View File
@@ -0,0 +1,19 @@
use axum::extract::State;
use conduwuit::Result;
use ruma::api::federation::edutypes::get_edutypes;
use crate::Ruma;
/// # `GET /_matrix/federation/v1/edutypes`
///
/// Lists EDU types we wish to receive
pub(crate) async fn get_edutypes_route(
State(services): State<crate::State>,
_body: Ruma<get_edutypes::unstable::Request>,
) -> Result<get_edutypes::unstable::Response> {
Ok(get_edutypes::unstable::Response {
typing: services.config.allow_incoming_typing,
presence: services.config.allow_incoming_presence,
receipt: services.config.allow_incoming_read_receipts,
})
}
+8 -5
View File
@@ -10,7 +10,6 @@
use ruma::{
CanonicalJsonValue, OwnedUserId, UserId,
api::{client::error::ErrorKind, federation::membership::create_invite},
events::room::member::{MembershipState, RoomMemberEventContent},
serde::JsonObject,
};
@@ -133,17 +132,21 @@ pub(crate) async fn create_invite_route(
services
.rooms
.state_cache
.update_membership(
&body.room_id,
.mark_as_invited(
&recipient_user,
RoomMemberEventContent::new(MembershipState::Invite),
&body.room_id,
sender_user,
Some(invite_state),
body.via.clone(),
true,
)
.await?;
services
.rooms
.state_cache
.update_joined_count(&body.room_id)
.await;
for appservice in services.appservice.read().await.values() {
if appservice.is_user_match(&recipient_user) {
services
+15 -4
View File
@@ -1,6 +1,6 @@
use axum::extract::State;
use conduwuit::{
Err, Error, Result, debug_info, matrix::pdu::PduBuilder, utils::IterStream, warn,
Err, Error, Result, debug_info, info, matrix::pdu::PduBuilder, utils::IterStream, warn,
};
use conduwuit_service::Services;
use futures::StreamExt;
@@ -22,6 +22,7 @@
/// # `GET /_matrix/federation/v1/make_join/{roomId}/{userId}`
///
/// Creates a join template.
#[tracing::instrument(skip_all, fields(room_id = %body.room_id, user_id = %body.user_id, origin = %body.origin()))]
pub(crate) async fn create_join_event_template_route(
State(services): State<crate::State>,
body: Ruma<prepare_join_event::v1::Request>,
@@ -72,11 +73,16 @@ pub(crate) async fn create_join_event_template_route(
}
let state_lock = services.rooms.state.mutex.lock(&body.room_id).await;
let is_invited = services
.rooms
.state_cache
.is_invited(&body.user_id, &body.room_id)
.await;
let join_authorized_via_users_server: Option<OwnedUserId> = {
use RoomVersionId::*;
if matches!(room_version_id, V1 | V2 | V3 | V4 | V5 | V6 | V7) {
// room version does not support restricted join rules
if matches!(room_version_id, V1 | V2 | V3 | V4 | V5 | V6 | V7) || is_invited {
// room version does not support restricted join rules, or the user is currently
// already invited
None
} else if user_can_perform_restricted_join(
&services,
@@ -103,6 +109,10 @@ pub(crate) async fn create_join_event_template_route(
.await
.map(ToOwned::to_owned)
else {
info!(
"No local user is able to authorize the join of {} into {}",
&body.user_id, &body.room_id
);
return Err!(Request(UnableToGrantJoin(
"No user on this server is able to assist in joining."
)));
@@ -167,6 +177,7 @@ pub(crate) async fn user_can_perform_restricted_join(
)
.await
else {
// No join rules means there's nothing to authorise (defaults to invite)
return Ok(false);
};
+2
View File
@@ -1,4 +1,5 @@
pub(super) mod backfill;
pub(super) mod edutypes;
pub(super) mod event;
pub(super) mod event_auth;
pub(super) mod get_missing_events;
@@ -23,6 +24,7 @@
pub(super) mod well_known;
pub(super) use backfill::*;
pub(super) use edutypes::*;
pub(super) use event::*;
pub(super) use event_auth::*;
pub(super) use get_missing_events::*;
+1 -1
View File
@@ -92,7 +92,7 @@ ruma.workspace = true
sanitize-filename.workspace = true
serde_json.workspace = true
serde_regex.workspace = true
serde_yml.workspace = true
serde-saphyr.workspace = true
serde.workspace = true
smallvec.workspace = true
smallstr.workspace = true
+32
View File
@@ -1128,6 +1128,23 @@ pub struct Config {
#[serde(default = "true_fn")]
pub rocksdb_bottommost_compression: bool,
/// Compression algorithm for RocksDB's Write-Ahead-Log (WAL).
///
/// At present, only ZSTD compression is supported by RocksDB for WAL
/// compression. Enabling this can reduce WAL size at the expense of some
/// CPU usage during writes.
///
/// The options are:
/// - "none" = No compression
/// - "zstd" = ZSTD compression
///
/// For more information on WAL compression, see:
/// https://github.com/facebook/rocksdb/wiki/WAL-Compression
///
/// default: "zstd"
#[serde(default = "default_rocksdb_wal_compression")]
pub rocksdb_wal_compression: String,
/// Database recovery mode (for RocksDB WAL corruption).
///
/// Use this option when the server reports corruption and refuses to start.
@@ -1710,6 +1727,19 @@ pub struct Config {
#[serde(default)]
pub block_non_admin_invites: bool,
/// Enable or disable making requests to MSC4284 Policy Servers.
/// It is recommended you keep this enabled unless you experience frequent
/// connectivity issues, such as in a restricted networking environment.
#[serde(default = "true_fn")]
pub enable_msc4284_policy_servers: bool,
/// Enable running locally generated events through configured MSC4284
/// policy servers. You may wish to disable this if your server is
/// single-user for a slight speed benefit in some rooms, but otherwise
/// should leave it enabled.
#[serde(default = "true_fn")]
pub policy_server_check_own_events: bool,
/// Allow admins to enter commands in rooms other than "#admins" (admin
/// room) by prefixing your message with "\!admin" or "\\!admin" followed up
/// a normal continuwuity admin command. The reply will be publicly visible
@@ -2441,6 +2471,8 @@ fn default_rocksdb_compression_algo() -> String {
.to_owned()
}
fn default_rocksdb_wal_compression() -> String { "zstd".to_owned() }
/// Default RocksDB compression level is 32767, which is internally read by
/// RocksDB as the default magic number and translated to the library's default
/// compression level as they all differ. See their `kDefaultCompressionLevel`.
+3 -1
View File
@@ -83,7 +83,9 @@ pub enum Error {
#[error(transparent)]
TypedHeader(#[from] axum_extra::typed_header::TypedHeaderRejection),
#[error(transparent)]
Yaml(#[from] serde_yml::Error),
YamlDe(#[from] serde_saphyr::Error),
#[error(transparent)]
YamlSer(#[from] serde_saphyr::ser_error::Error),
// ruma/conduwuit
#[error("Arithmetic operation failed: {0}")]
+7 -2
View File
@@ -5,13 +5,17 @@
use std::{collections::BTreeMap, sync::OnceLock};
use crate::{SyncMutex, utils::exchange};
use crate::utils::exchange;
/// Raw capture of rustc flags used to build each crate in the project. Informed
/// by rustc_flags_capture macro (one in each crate's mod.rs). This is
/// done during static initialization which is why it's mutex-protected and pub.
/// Should not be written to by anything other than our macro.
pub static FLAGS: SyncMutex<BTreeMap<&str, &[&str]>> = SyncMutex::new(BTreeMap::new());
///
/// We specifically use a std mutex here because parking_lot cannot be used
/// after thread local storage is destroyed on MacOS.
pub static FLAGS: std::sync::Mutex<BTreeMap<&str, &[&str]>> =
std::sync::Mutex::new(BTreeMap::new());
/// Processed list of enabled features across all project crates. This is
/// generated from the data in FLAGS.
@@ -24,6 +28,7 @@ fn init_features() -> Vec<&'static str> {
let mut features = Vec::new();
FLAGS
.lock()
.expect("locked")
.iter()
.for_each(|(_, flags)| append_features(&mut features, flags));
+341 -100
View File
@@ -200,11 +200,15 @@ pub async fn auth_check<E, F, Fut>(
if incoming_event.room_id().is_some() {
let Some(room_id_server_name) = incoming_event.room_id().unwrap().server_name()
else {
warn!("room ID has no servername");
warn!("legacy room ID has no server name");
return Ok(false);
};
if room_id_server_name != sender.server_name() {
warn!("servername of room ID does not match servername of sender");
warn!(
expected = %sender.server_name(),
received = %room_id_server_name,
"server name of legacy room ID does not match server name of sender"
);
return Ok(false);
}
}
@@ -215,12 +219,12 @@ pub async fn auth_check<E, F, Fut>(
.room_version
.is_some_and(|v| v.deserialize().is_err())
{
warn!("invalid room version found in m.room.create event");
warn!("unsupported room version found in m.room.create event");
return Ok(false);
}
if room_version.room_ids_as_hashes && incoming_event.room_id().is_some() {
warn!("room create event incorrectly claims a room ID");
warn!("room create event incorrectly claims to have a room ID when it should not");
return Ok(false);
}
@@ -229,7 +233,7 @@ pub async fn auth_check<E, F, Fut>(
{
// If content has no creator field, reject
if content.creator.is_none() {
warn!("no creator field found in m.room.create content");
warn!("m.room.create event incorrectly omits 'creator' field");
return Ok(false);
}
}
@@ -282,16 +286,19 @@ pub async fn auth_check<E, F, Fut>(
.room_version
.is_some_and(|v| v.deserialize().is_err())
{
warn!("invalid room version found in m.room.create event");
warn!(
create_event_id = %room_create_event.event_id(),
"unsupported room version found in m.room.create event"
);
return Ok(false);
}
let expected_room_id = room_create_event.room_id_or_hash();
if incoming_event.room_id().unwrap() != expected_room_id {
if incoming_event.room_id().expect("event must have a room ID") != expected_room_id {
warn!(
expected = %expected_room_id,
received = %incoming_event.room_id().unwrap(),
"room_id of incoming event ({}) does not match room_id of m.room.create event ({})",
"room_id of incoming event ({}) does not match that of the m.room.create event ({})",
incoming_event.room_id().unwrap(),
expected_room_id,
);
@@ -304,12 +311,15 @@ pub async fn auth_check<E, F, Fut>(
.auth_events()
.any(|id| id == room_create_event.event_id());
if room_version.room_ids_as_hashes && claims_create_event {
warn!("m.room.create event incorrectly found in auth events");
warn!("event incorrectly references m.room.create event in auth events");
return Ok(false);
} else if !room_version.room_ids_as_hashes && !claims_create_event {
// If the create event is not referenced in the event's auth events, and this is
// a v11 room, reject
warn!("no m.room.create event found in auth events");
warn!(
missing = %room_create_event.event_id(),
"event incorrectly did not reference an m.room.create in its auth events"
);
return Ok(false);
}
@@ -318,7 +328,7 @@ pub async fn auth_check<E, F, Fut>(
warn!(
expected = %expected_room_id,
received = %pe.room_id().unwrap(),
"room_id of power levels event does not match room_id of m.room.create event"
"room_id of referenced power levels event does not match that of the m.room.create event"
);
return Ok(false);
}
@@ -332,8 +342,9 @@ pub async fn auth_check<E, F, Fut>(
&& room_create_event.sender().server_name() != incoming_event.sender().server_name()
{
warn!(
"room is not federated and event's sender domain does not match create event's \
sender domain"
sender = %incoming_event.sender(),
create_sender = %room_create_event.sender(),
"room is not federated and event's sender domain does not match create event's sender domain"
);
return Ok(false);
}
@@ -416,7 +427,6 @@ pub async fn auth_check<E, F, Fut>(
&user_for_join_auth_membership,
&room_create_event,
)? {
warn!("membership change not valid for some reason");
return Ok(false);
}
@@ -429,7 +439,7 @@ pub async fn auth_check<E, F, Fut>(
let sender_member_event = match sender_member_event {
| Some(mem) => mem,
| None => {
warn!("sender not found in room");
warn!("sender has no membership event");
return Ok(false);
},
};
@@ -440,7 +450,7 @@ pub async fn auth_check<E, F, Fut>(
!= expected_room_id
{
warn!(
"room_id of incoming event ({}) does not match room_id of m.room.create event ({})",
"room_id of incoming event ({}) does not match that of the m.room.create event ({})",
sender_member_event
.room_id()
.expect("event must have a room ID"),
@@ -453,8 +463,7 @@ pub async fn auth_check<E, F, Fut>(
from_json_str(sender_member_event.content().get())?;
let Some(membership_state) = sender_membership_event_content.membership else {
warn!(
sender_membership_event_content = format!("{sender_membership_event_content:?}"),
event_id = format!("{}", incoming_event.event_id()),
?sender_membership_event_content,
"Sender membership event content missing membership field"
);
return Err(Error::InvalidPdu("Missing membership field".to_owned()));
@@ -462,7 +471,11 @@ pub async fn auth_check<E, F, Fut>(
let membership_state = membership_state.deserialize()?;
if !matches!(membership_state, MembershipState::Join) {
warn!("sender's membership is not join");
warn!(
%sender,
?membership_state,
"sender cannot send events without being joined to the room"
);
return Ok(false);
}
@@ -522,7 +535,12 @@ pub async fn auth_check<E, F, Fut>(
};
if sender_power_level < invite_level {
warn!("sender's cannot send invites in this room");
warn!(
%sender,
has=?sender_power_level,
required=?invite_level,
"sender cannot send invites in this room"
);
return Ok(false);
}
@@ -534,7 +552,11 @@ pub async fn auth_check<E, F, Fut>(
// level, reject If the event has a state_key that starts with an @ and does
// not match the sender, reject.
if !can_send_event(incoming_event, power_levels_event.as_ref(), sender_power_level) {
warn!("user cannot send event");
warn!(
%sender,
event_type=?incoming_event.kind(),
"sender cannot send event"
);
return Ok(false);
}
@@ -579,6 +601,12 @@ pub async fn auth_check<E, F, Fut>(
};
if !check_redaction(room_version, incoming_event, sender_power_level, redact_level)? {
warn!(
%sender,
?sender_power_level,
?redact_level,
"redaction event was not allowed"
);
return Ok(false);
}
}
@@ -587,15 +615,21 @@ pub async fn auth_check<E, F, Fut>(
Ok(true)
}
fn is_creator<EV>(v: &RoomVersion, c: &BTreeSet<OwnedUserId>, ce: &EV, user_id: &UserId) -> bool
fn is_creator<EV>(
v: &RoomVersion,
c: &BTreeSet<OwnedUserId>,
ce: &EV,
user_id: &UserId,
have_pls: bool,
) -> bool
where
EV: Event + Send + Sync,
{
if v.explicitly_privilege_room_creators {
c.contains(user_id)
} else if v.use_room_create_sender {
} else if v.use_room_create_sender && !have_pls {
ce.sender() == user_id
} else {
} else if !have_pls {
#[allow(deprecated)]
let creator = from_json_str::<RoomCreateEventContent>(ce.content().get())
.unwrap()
@@ -604,6 +638,8 @@ fn is_creator<EV>(v: &RoomVersion, c: &BTreeSet<OwnedUserId>, ce: &EV, user_id:
.unwrap();
creator == user_id
} else {
false
}
}
@@ -696,10 +732,11 @@ struct GetThirdPartyInvite {
}
trace!(?creators, "creators for room");
let mut join_rules = JoinRule::Invite;
if let Some(jr) = &join_rules_event {
join_rules = from_json_str::<RoomJoinRulesEventContent>(jr.content().get())?.join_rule;
}
let join_rules = if let Some(jr) = &join_rules_event {
from_json_str::<RoomJoinRulesEventContent>(jr.content().get())?.join_rule
} else {
JoinRule::Invite
};
let power_levels_event_id = power_levels_event.as_ref().map(Event::event_id);
let sender_membership_event_id = sender_membership_event.as_ref().map(Event::event_id);
@@ -725,8 +762,13 @@ struct GetThirdPartyInvite {
(int!(0), int!(0))
};
let user_joined = user_for_join_auth_membership == &MembershipState::Join;
let okay_power = is_creator(room_version, &creators, create_room, user_for_join_auth)
|| auth_user_pl >= invite_level;
let okay_power = is_creator(
room_version,
&creators,
create_room,
user_for_join_auth,
power_levels_event.as_ref().is_some(),
) || auth_user_pl >= invite_level;
trace!(
auth_user_pl=?auth_user_pl,
invite_level=?invite_level,
@@ -741,8 +783,20 @@ struct GetThirdPartyInvite {
trace!("No auth user given for join auth");
false
};
let sender_creator = is_creator(room_version, &creators, create_room, sender);
let target_creator = is_creator(room_version, &creators, create_room, target_user);
let sender_creator = is_creator(
room_version,
&creators,
create_room,
sender,
power_levels_event.as_ref().is_some(),
);
let target_creator = is_creator(
room_version,
&creators,
create_room,
target_user,
power_levels_event.as_ref().is_some(),
);
Ok(match target_membership {
| MembershipState::Join => {
@@ -759,7 +813,7 @@ struct GetThirdPartyInvite {
if prev_event_is_create_event && no_more_prev_events {
trace!(
sender = %sender,
%sender,
target_user = %target_user,
?sender_creator,
?target_creator,
@@ -779,22 +833,33 @@ struct GetThirdPartyInvite {
);
if sender != target_user {
// If the sender does not match state_key, reject.
warn!("Can't make other user join");
warn!(
%sender,
target_user = %target_user,
"sender cannot join on behalf of another user"
);
false
} else if target_user_current_membership == MembershipState::Ban {
// If the sender is banned, reject.
warn!(?target_user_membership_event_id, "Banned user can't join");
warn!(
%sender,
membership_event_id = ?target_user_membership_event_id,
"sender cannot join as they are banned from the room"
);
false
} else {
match join_rules {
| JoinRule::Invite =>
if !membership_allows_join {
warn!(
membership=?target_user_current_membership,
"Join rule is invite but membership does not allow join"
%sender,
membership_event_id = ?target_user_membership_event_id,
membership = ?target_user_current_membership,
"sender cannot join as they are not invited to the invite-only room"
);
false
} else {
trace!(sender=%sender, "sender is invited to room, allowing join");
true
},
| JoinRule::Knock if !room_version.allow_knocking => {
@@ -804,11 +869,14 @@ struct GetThirdPartyInvite {
| JoinRule::Knock =>
if !membership_allows_join {
warn!(
%sender,
membership_event_id = ?target_user_membership_event_id,
membership=?target_user_current_membership,
"Join rule is knock but membership does not allow join"
"sender cannot join a knock room without being invited or already joined"
);
false
} else {
trace!(sender=%sender, "sender is invited or already joined to room, allowing join");
true
},
| JoinRule::KnockRestricted(_) if !room_version.knock_restricted_join_rule =>
@@ -820,33 +888,56 @@ struct GetThirdPartyInvite {
},
| JoinRule::KnockRestricted(_) => {
if membership_allows_join || user_for_join_auth_is_valid {
trace!(
%sender,
%membership_allows_join,
%user_for_join_auth_is_valid,
"sender is invited, already joined to, or authorised to join the room, allowing join"
);
true
} else {
warn!(
%sender,
membership_event_id = ?target_user_membership_event_id,
membership=?target_user_current_membership,
"Join rule is a restricted one, but no valid authorising user \
was given and the sender's current membership does not permit \
a join transition"
%user_for_join_auth_is_valid,
?user_for_join_auth,
"sender cannot join as they are not invited nor already joined to the room, nor was a \
valid authorising user given to permit the join"
);
false
}
},
| JoinRule::Restricted(_) =>
| JoinRule::Restricted(_) => {
if membership_allows_join || user_for_join_auth_is_valid {
trace!(
%sender,
%membership_allows_join,
%user_for_join_auth_is_valid,
"sender is invited, already joined to, or authorised to join the room, allowing join"
);
true
} else {
warn!(
"Join rule is a restricted one but no valid authorising user \
was given"
%sender,
membership_event_id = ?target_user_membership_event_id,
membership=?target_user_current_membership,
%user_for_join_auth_is_valid,
?user_for_join_auth,
"sender cannot join as they are not invited nor already joined to the room, nor was a \
valid authorising user given to permit the join"
);
false
},
| JoinRule::Public => true,
}
},
| JoinRule::Public => {
trace!(%sender, "join rule is public, allowing join");
true
},
| _ => {
warn!(
join_rule=?join_rules,
membership=?target_user_current_membership,
"Unknown join rule doesn't allow joining, or the rule's conditions were not met"
"Join rule is unknown, or the rule's conditions were not met"
);
false
},
@@ -873,16 +964,23 @@ struct GetThirdPartyInvite {
}
allow
},
| _ => {
if !sender_is_joined
|| target_user_current_membership == MembershipState::Join
|| target_user_current_membership == MembershipState::Ban
{
| _ =>
if !sender_is_joined {
warn!(
%sender,
?sender_membership_event_id,
?sender_membership,
"sender cannot produce an invite without being joined to the room",
);
false
} else if matches!(
target_user_current_membership,
MembershipState::Join | MembershipState::Ban
) {
warn!(
?target_user_membership_event_id,
?sender_membership_event_id,
"Can't invite user if sender not joined or the user is currently \
joined or banned",
?target_user_current_membership,
"cannot invite a user who is banned or already joined",
);
false
} else {
@@ -892,56 +990,124 @@ struct GetThirdPartyInvite {
.is_some();
if !allow {
warn!(
?target_user_membership_event_id,
?power_levels_event_id,
"User does not have enough power to invite",
%sender,
has=?sender_power,
required=?power_levels.invite,
"sender does not have enough power to produce invites",
);
}
trace!(
%sender,
?sender_membership_event_id,
?sender_membership,
?target_user_membership_event_id,
?target_user_current_membership,
sender_pl=?sender_power,
required_pl=?power_levels.invite,
"allowing invite"
);
allow
}
},
},
}
},
| MembershipState::Leave =>
| MembershipState::Leave => {
let can_unban = if target_user_current_membership == MembershipState::Ban {
sender_creator || sender_power.filter(|&p| p >= &power_levels.ban).is_some()
} else {
true
};
let can_kick = if !matches!(
target_user_current_membership,
MembershipState::Ban | MembershipState::Leave
) {
if sender_creator {
// sender is a creator
true
} else if sender_power.filter(|&p| p >= &power_levels.kick).is_none() {
// sender lacks kick power level
false
} else if let Some(sp) = sender_power {
if let Some(tp) = target_power {
// sender must have more power than target
sp > tp
} else {
// target has default power level
true
}
} else {
// sender has default power level
false
}
} else {
true
};
if sender == target_user {
let allow = target_user_current_membership == MembershipState::Join
|| target_user_current_membership == MembershipState::Invite
|| target_user_current_membership == MembershipState::Knock;
// self-leave
// let allow = target_user_current_membership == MembershipState::Join
// || target_user_current_membership == MembershipState::Invite
// || target_user_current_membership == MembershipState::Knock;
let allow = matches!(
target_user_current_membership,
MembershipState::Join | MembershipState::Invite | MembershipState::Knock
);
if !allow {
warn!(
?target_user_membership_event_id,
?target_user_current_membership,
"Can't leave if sender is not already invited, knocked, or joined"
%sender,
current_membership_event_id=?target_user_membership_event_id,
current_membership=?target_user_current_membership,
"sender cannot leave as they are not already knocking on, invited to, or joined to the room"
);
}
trace!(sender=%sender, "allowing leave");
allow
} else if !sender_is_joined
|| target_user_current_membership == MembershipState::Ban
&& (sender_creator
|| sender_power.filter(|&p| p < &power_levels.ban).is_some())
{
} else if !sender_is_joined {
warn!(
?target_user_membership_event_id,
%sender,
?sender_membership_event_id,
"Can't kick if sender not joined or user is already banned",
"sender cannot kick another user as they are not joined to the room",
);
false
} else if !(can_unban && can_kick) {
// If the target is banned, only a room creator or someone with ban power
// level can unban them
warn!(
%sender,
?target_user_membership_event_id,
?power_levels_event_id,
"sender lacks the power level required to unban users",
);
false
} else if !can_kick {
warn!(
%sender,
%target_user,
?target_user_membership_event_id,
?target_user_current_membership,
?power_levels_event_id,
"sender does not have enough power to kick the target",
);
false
} else {
let allow = sender_creator
|| (sender_power.filter(|&p| p >= &power_levels.kick).is_some()
&& target_power < sender_power);
if !allow {
warn!(
?target_user_membership_event_id,
?power_levels_event_id,
"User does not have enough power to kick",
);
}
allow
},
trace!(
%sender,
%target_user,
?target_user_membership_event_id,
?target_user_current_membership,
sender_pl=?sender_power,
target_pl=?target_power,
required_pl=?power_levels.kick,
"allowing kick/unban",
);
true
}
},
| MembershipState::Ban =>
if !sender_is_joined {
warn!(?sender_membership_event_id, "Can't ban user if sender is not joined");
warn!(
%sender,
?sender_membership_event_id,
"sender cannot ban another user as they are not joined to the room",
);
false
} else {
let allow = sender_creator
@@ -949,9 +1115,11 @@ struct GetThirdPartyInvite {
&& target_power < sender_power);
if !allow {
warn!(
%sender,
%target_user,
?target_user_membership_event_id,
?power_levels_event_id,
"User does not have enough power to ban",
"sender does not have enough power to ban the target",
);
}
allow
@@ -977,9 +1145,9 @@ struct GetThirdPartyInvite {
} else if sender != target_user {
// 3. If `sender` does not match `state_key`, reject.
warn!(
?sender,
?target_user,
"Can't make another user knock, sender did not match target"
%sender,
%target_user,
"sender cannot knock on behalf of another user",
);
false
} else if matches!(
@@ -991,15 +1159,25 @@ struct GetThirdPartyInvite {
// 5. Otherwise, reject.
warn!(
?target_user_membership_event_id,
?sender_membership,
"Knocking with a membership state of ban, invite or join is invalid",
);
false
} else {
trace!(%sender, "allowing knock");
true
}
},
| _ => {
warn!("Unknown membership transition");
warn!(
%sender,
?target_membership,
%target_user,
%target_user_current_membership,
"Unknown or invalid membership transition {} -> {}",
target_user_current_membership,
target_membership
);
false
},
})
@@ -1029,6 +1207,13 @@ fn can_send_event(event: &impl Event, ple: Option<&impl Event>, user_level: Int)
if event.state_key().is_some_and(|k| k.starts_with('@'))
&& event.state_key() != Some(event.sender().as_str())
{
warn!(
%user_level,
required=?event_type_power_level,
state_key=?event.state_key(),
sender=%event.sender(),
"state_key starts with @ but does not match sender",
);
return false; // permission required to post in this room
}
@@ -1113,7 +1298,14 @@ fn check_power_levels(
// If the current value is equal to the sender's current power level, reject
if user != power_event.sender() && old_level == Some(&user_level) {
warn!("m.room.power_level cannot remove ops == to own");
warn!(
?old_level,
?new_level,
?user,
%user_level,
sender=%power_event.sender(),
"cannot alter the power level of a user with the same power level as sender's own"
);
return Some(false); // cannot remove ops level == to own
}
@@ -1121,8 +1313,26 @@ fn check_power_levels(
// If the new value is higher than the sender's current power level, reject
let old_level_too_big = old_level > Some(&user_level);
let new_level_too_big = new_level > Some(&user_level);
if old_level_too_big || new_level_too_big {
warn!("m.room.power_level failed to add ops > than own");
if old_level_too_big {
warn!(
?old_level,
?new_level,
?user,
%user_level,
sender=%power_event.sender(),
"cannot alter the power level of a user with a higher power level than sender's own"
);
return Some(false); // cannot add ops greater than own
}
if new_level_too_big {
warn!(
?old_level,
?new_level,
?user,
%user_level,
sender=%power_event.sender(),
"cannot set the power level of a user to a level higher than sender's own"
);
return Some(false); // cannot add ops greater than own
}
}
@@ -1139,8 +1349,26 @@ fn check_power_levels(
// If the new value is higher than the sender's current power level, reject
let old_level_too_big = old_level > Some(&user_level);
let new_level_too_big = new_level > Some(&user_level);
if old_level_too_big || new_level_too_big {
warn!("m.room.power_level failed to add ops > than own");
if old_level_too_big {
warn!(
?old_level,
?new_level,
?ev_type,
%user_level,
sender=%power_event.sender(),
"cannot alter the power level of an event with a higher power level than sender's own"
);
return Some(false); // cannot add ops greater than own
}
if new_level_too_big {
warn!(
?old_level,
?new_level,
?ev_type,
%user_level,
sender=%power_event.sender(),
"cannot set the power level of an event to a level higher than sender's own"
);
return Some(false); // cannot add ops greater than own
}
}
@@ -1155,7 +1383,13 @@ fn check_power_levels(
let old_level_too_big = old_level > user_level;
let new_level_too_big = new_level > user_level;
if old_level_too_big || new_level_too_big {
warn!("m.room.power_level failed to add ops > than own");
warn!(
?old_level,
?new_level,
%user_level,
sender=%power_event.sender(),
"cannot alter the power level of notifications greater than sender's own"
);
return Some(false); // cannot add ops greater than own
}
}
@@ -1179,7 +1413,14 @@ fn check_power_levels(
let new_level_too_big = new_lvl > user_level;
if old_level_too_big || new_level_too_big {
warn!("cannot add ops > than own");
warn!(
?old_lvl,
?new_lvl,
%user_level,
sender=%power_event.sender(),
action=%lvl_name,
"cannot alter the power level of action greater than sender's own",
);
return Some(false);
}
}
+24 -29
View File
@@ -101,40 +101,40 @@ pub async fn resolve<'a, Pdu, Sets, SetIter, Hasher, Fetch, FetchFut, Exists, Ex
debug!(version = ?stateres_version, "State resolution starting");
// Split non-conflicting and conflicting state
let (clean, conflicting) = separate(state_sets.into_iter());
let (unconflicted, conflicting) = separate(state_sets.into_iter());
debug!(count = clean.len(), "non-conflicting events");
trace!(map = ?clean, "non-conflicting events");
debug!(count = unconflicted.len(), "non-conflicting events");
trace!(map = ?unconflicted, "non-conflicting events");
if conflicting.is_empty() {
debug!("no conflicting state found");
return Ok(clean);
return Ok(unconflicted);
}
debug!(count = conflicting.len(), "conflicting events");
trace!(map = ?conflicting, "conflicting events");
let conflicted_state_subgraph: HashSet<_> = match stateres_version {
| StateResolutionVersion::V2_1 =>
calculate_conflicted_subgraph(&conflicting, event_fetch)
let (conflicted_state_subgraph, initial_state) =
if stateres_version == StateResolutionVersion::V2_1 {
let csg = calculate_conflicted_subgraph(&conflicting, event_fetch)
.await
.ok_or_else(|| {
Error::InvalidPdu("Failed to calculate conflicted subgraph".to_owned())
})?,
| _ => HashSet::new(),
};
debug!(count = conflicted_state_subgraph.len(), "conflicted subgraph");
trace!(set = ?conflicted_state_subgraph, "conflicted subgraph");
let conflicting_values = conflicting.into_values().flatten().stream();
})?;
debug!(count = csg.len(), "conflicted subgraph");
trace!(set = ?csg, "conflicted subgraph");
(csg, HashMap::new())
} else {
(HashSet::new(), unconflicted.clone())
};
// `all_conflicted` contains unique items
// synapse says `full_set = {eid for eid in full_conflicted_set if eid in
// event_map}`
// Hydra: Also consider the conflicted state subgraph
let all_conflicted: HashSet<_> = get_auth_chain_diff(auth_chain_sets)
.chain(conflicting_values)
.chain(conflicted_state_subgraph.into_iter().stream())
.chain(conflicting.into_values().flatten().stream())
.broad_filter_map(async |id| event_exists(id.clone()).await.then_some(id))
.chain(conflicted_state_subgraph.into_iter().stream())
.collect()
.await;
@@ -171,7 +171,7 @@ pub async fn resolve<'a, Pdu, Sets, SetIter, Hasher, Fetch, FetchFut, Exists, Ex
&room_version,
&stateres_version,
sorted_control_levels.iter().stream().map(AsRef::as_ref),
clean.clone(),
initial_state,
&event_fetch,
)
.await?;
@@ -201,7 +201,7 @@ pub async fn resolve<'a, Pdu, Sets, SetIter, Hasher, Fetch, FetchFut, Exists, Ex
let power_levels_ty_sk = (StateEventType::RoomPowerLevels, StateKey::new());
let power_event = resolved_control.get(&power_levels_ty_sk);
debug!(event_id = ?power_event, "power event");
trace!(event_id = ?power_event, "power event");
let sorted_left_events =
mainline_sort(&events_to_resolve, power_event.cloned(), &event_fetch).await?;
@@ -212,19 +212,13 @@ pub async fn resolve<'a, Pdu, Sets, SetIter, Hasher, Fetch, FetchFut, Exists, Ex
&room_version,
&stateres_version,
sorted_left_events.iter().stream().map(AsRef::as_ref),
resolved_control.clone(), // The control events are added to the final resolved state
resolved_control, // The control events are added to the final resolved state
&event_fetch,
)
.await?;
// Add unconflicted state to the resolved state
// We priorities the unconflicting state
resolved_state.extend(clean);
if stateres_version == StateResolutionVersion::V2_1 {
resolved_state.extend(resolved_control);
// TODO(hydra): this feels disgusting and wrong but it allows
// the state to resolve properly?
}
// Ensure unconflicting state is in the final state
resolved_state.extend(unconflicted);
debug!("state resolution finished");
trace!( map = ?resolved_state, "final resolved state" );
@@ -427,8 +421,8 @@ async fn reverse_topological_power_sort<E, F, Fut>(
/// `key_fn` is used as to obtain the power level and age of an event for
/// breaking ties (together with the event ID).
#[tracing::instrument(level = "debug", skip_all)]
pub async fn lexicographical_topological_sort<Id, F, Fut, Hasher>(
graph: &HashMap<Id, HashSet<Id, Hasher>>,
pub async fn lexicographical_topological_sort<Id, F, Fut, Hasher, S>(
graph: &HashMap<Id, HashSet<Id, Hasher>, S>,
key_fn: &F,
) -> Result<Vec<Id>>
where
@@ -436,6 +430,7 @@ pub async fn lexicographical_topological_sort<Id, F, Fut, Hasher>(
Fut: Future<Output = Result<(Int, MilliSecondsSinceUnixEpoch)>> + Send,
Id: Borrow<EventId> + Clone + Eq + Hash + Ord + Send + Sync,
Hasher: BuildHasher + Default + Clone + Send + Sync,
S: BuildHasher + Clone + Send + Sync,
{
#[derive(PartialEq, Eq)]
struct TieBreaker<'a, Id> {
+2 -2
View File
@@ -36,7 +36,7 @@ fn map_ok_or<U, F>(
) -> MapOkOrElse<Self, impl FnOnce(Self::Ok) -> U, impl FnOnce(Self::Error) -> U>
where
F: FnOnce(Self::Ok) -> U,
Self: Send + Sized;
Self: Sized;
fn ok(
self,
@@ -100,7 +100,7 @@ fn map_ok_or<U, F>(
) -> MapOkOrElse<Self, impl FnOnce(Self::Ok) -> U, impl FnOnce(Self::Error) -> U>
where
F: FnOnce(Self::Ok) -> U,
Self: Send + Sized,
Self: Sized,
{
self.map_ok_or_else(|_| default, f)
}
+19
View File
@@ -276,3 +276,22 @@ async fn set_intersection_sorted_stream2() {
.await;
assert!(r.eq(&["ccc", "ggg", "iii"]));
}
#[test]
fn is_within_bounds() {
use std::time::{Duration, SystemTime};
use utils::time::{TimeDirection, is_within_bounds};
let now = SystemTime::now();
let yesterday = now - Duration::from_secs(86400);
assert!(is_within_bounds(yesterday, now, TimeDirection::Before));
assert!(!is_within_bounds(yesterday, now, TimeDirection::After));
let tomorrow = now + Duration::from_secs(86400);
assert!(is_within_bounds(tomorrow, now, TimeDirection::After));
assert!(!is_within_bounds(tomorrow, now, TimeDirection::Before));
assert!(is_within_bounds(now, now, TimeDirection::Before));
assert!(is_within_bounds(now, now, TimeDirection::After));
}
+21
View File
@@ -126,3 +126,24 @@ pub enum Unit {
Micros(u128),
Nanos(u128),
}
#[derive(Eq, PartialEq, Clone, Copy, Debug)]
pub enum TimeDirection {
Before,
After,
}
/// Checks if `item_time` is before or after `time_boundary`.
/// If both times are the same, it will return true for both directions, as the
/// matching is inclusive.
#[must_use]
pub fn is_within_bounds(
item_time: SystemTime,
time_boundary: SystemTime,
direction: TimeDirection,
) -> bool {
match direction {
| TimeDirection::Before => item_time <= time_boundary,
| TimeDirection::After => item_time >= time_boundary,
}
}
+18 -2
View File
@@ -1,7 +1,9 @@
use std::{cmp, convert::TryFrom};
use conduwuit::{Config, Result, utils};
use rocksdb::{Cache, DBRecoveryMode, Env, LogLevel, Options, statistics::StatsLevel};
use conduwuit::{Config, Result, utils, warn};
use rocksdb::{
Cache, DBCompressionType, DBRecoveryMode, Env, LogLevel, Options, statistics::StatsLevel,
};
use super::{cf_opts::cache_size_f64, logger::handle as handle_log};
@@ -58,6 +60,20 @@ pub(crate) fn db_options(config: &Config, env: &Env, row_cache: &Cache) -> Resul
opts.set_max_total_wal_size(1024 * 1024 * 512);
opts.set_writable_file_max_buffer_size(1024 * 1024 * 2);
// WAL compression
let wal_compression = match config.rocksdb_wal_compression.as_ref() {
| "zstd" => DBCompressionType::Zstd,
| "none" => DBCompressionType::None,
| value => {
warn!(
"Invalid rocksdb_wal_compression value '{value}'. Supported values are 'none' \
or 'zstd'. Defaulting to 'none'."
);
DBCompressionType::None
},
};
opts.set_wal_compression_type(wal_compression);
// Misc
opts.set_disable_auto_compactions(!config.rocksdb_compaction);
opts.create_missing_column_families(true);
+1 -15
View File
@@ -10,7 +10,7 @@
use futures::{Stream, StreamExt, TryStreamExt};
use rocksdb::{DBPinnableSlice, ReadOptions};
use super::get::{cached_handle_from, handle_from};
use super::get::handle_from;
use crate::Handle;
pub trait Get<'a, K, S>
@@ -58,20 +58,6 @@ pub(crate) fn get_batch<'a, S, K>(
.try_flatten()
}
#[implement(super::Map)]
#[tracing::instrument(name = "batch_cached", level = "trace", skip_all)]
pub(crate) fn get_batch_cached<'a, I, K>(
&self,
keys: I,
) -> impl Iterator<Item = Result<Option<Handle<'_>>>> + Send + use<'_, I, K>
where
I: Iterator<Item = &'a K> + ExactSizeIterator + Send,
K: AsRef<[u8]> + Send + ?Sized + Sync + 'a,
{
self.get_batch_blocking_opts(keys, &self.cache_read_options)
.map(cached_handle_from)
}
#[implement(super::Map)]
#[tracing::instrument(name = "batch_blocking", level = "trace", skip_all)]
pub(crate) fn get_batch_blocking<'a, I, K>(
+11 -11
View File
@@ -184,7 +184,7 @@ fn spawn_one(
let handle = thread::Builder::new()
.name(WORKER_NAME.into())
.stack_size(WORKER_STACK_SIZE)
.spawn(move || self.worker(id, recv))?;
.spawn(move || self.worker(id, &recv))?;
workers.push(handle);
@@ -260,9 +260,9 @@ async fn execute(&self, queue: &Sender<Cmd>, cmd: Cmd) -> Result {
tid = ?thread::current().id(),
),
)]
fn worker(self: Arc<Self>, id: usize, recv: Receiver<Cmd>) {
fn worker(self: Arc<Self>, id: usize, recv: &Receiver<Cmd>) {
self.worker_init(id);
self.worker_loop(&recv);
self.worker_loop(recv);
}
#[implement(Pool)]
@@ -309,7 +309,7 @@ fn worker_loop(self: &Arc<Self>, recv: &Receiver<Cmd>) {
self.busy.fetch_add(1, Ordering::Relaxed);
while let Ok(cmd) = self.worker_wait(recv) {
self.worker_handle(cmd);
Pool::worker_handle(cmd);
}
}
@@ -331,11 +331,11 @@ fn worker_wait(self: &Arc<Self>, recv: &Receiver<Cmd>) -> Result<Cmd, RecvError>
}
#[implement(Pool)]
fn worker_handle(self: &Arc<Self>, cmd: Cmd) {
fn worker_handle(cmd: Cmd) {
match cmd {
| Cmd::Get(cmd) if cmd.key.len() == 1 => self.handle_get(cmd),
| Cmd::Get(cmd) => self.handle_batch(cmd),
| Cmd::Iter(cmd) => self.handle_iter(cmd),
| Cmd::Get(cmd) if cmd.key.len() == 1 => Pool::handle_get(cmd),
| Cmd::Get(cmd) => Pool::handle_batch(cmd),
| Cmd::Iter(cmd) => Pool::handle_iter(cmd),
}
}
@@ -346,7 +346,7 @@ fn worker_handle(self: &Arc<Self>, cmd: Cmd) {
skip_all,
fields(%cmd.map),
)]
fn handle_iter(&self, mut cmd: Seek) {
fn handle_iter(mut cmd: Seek) {
let chan = cmd.res.take().expect("missing result channel");
if chan.is_canceled() {
@@ -375,7 +375,7 @@ fn handle_iter(&self, mut cmd: Seek) {
keys = %cmd.key.len(),
),
)]
fn handle_batch(self: &Arc<Self>, mut cmd: Get) {
fn handle_batch(mut cmd: Get) {
debug_assert!(cmd.key.len() > 1, "should have more than one key");
debug_assert!(!cmd.key.iter().any(SmallVec::is_empty), "querying for empty key");
@@ -401,7 +401,7 @@ fn handle_batch(self: &Arc<Self>, mut cmd: Get) {
skip_all,
fields(%cmd.map),
)]
fn handle_get(&self, mut cmd: Get) {
fn handle_get(mut cmd: Get) {
debug_assert!(!cmd.key[0].is_empty(), "querying for empty key");
// Obtain the result channel.
+2 -2
View File
@@ -25,13 +25,13 @@ pub(super) fn refutable(mut item: ItemFn, _args: &[Meta]) -> Result<TokenStream>
};
let name = format!("_args_{i}");
*pat = Box::new(Pat::Ident(PatIdent {
**pat = Pat::Ident(PatIdent {
ident: Ident::new(&name, Span::call_site().into()),
attrs: Vec::new(),
by_ref: None,
mutability: None,
subpat: None,
}));
});
let field = fields.iter();
let refute = quote! {
+2 -2
View File
@@ -15,13 +15,13 @@ pub(super) fn flags_capture(args: TokenStream) -> TokenStream {
#[ctor]
fn _set_rustc_flags() {
conduwuit_core::info::rustc::FLAGS.lock().insert(#crate_name, &RUSTC_FLAGS);
conduwuit_core::info::rustc::FLAGS.lock().expect("locked").insert(#crate_name, &RUSTC_FLAGS);
}
// static strings have to be yanked on module unload
#[dtor]
fn _unset_rustc_flags() {
conduwuit_core::info::rustc::FLAGS.lock().remove(#crate_name);
conduwuit_core::info::rustc::FLAGS.lock().expect("locked").remove(#crate_name);
}
};
+4 -4
View File
@@ -62,7 +62,8 @@ standard = [
"media_thumbnail",
"systemd",
"url_preview",
"zstd_compression"
"zstd_compression",
"sentry_telemetry"
]
full = [
"standard",
@@ -129,7 +130,6 @@ perf_measurements = [
"dep:tracing-opentelemetry",
"dep:opentelemetry_sdk",
"dep:opentelemetry-otlp",
"dep:opentelemetry-jaeger-propagator",
"conduwuit-core/perf_measurements",
"conduwuit-core/sentry_telemetry",
]
@@ -156,6 +156,7 @@ sentry_telemetry = [
]
systemd = [
"conduwuit-router/systemd",
"conduwuit-service/systemd"
]
journald = [ # This is a stub on non-unix platforms
"dep:tracing-journald",
@@ -200,6 +201,7 @@ conduwuit-core.workspace = true
conduwuit-database.workspace = true
conduwuit-router.workspace = true
conduwuit-service.workspace = true
conduwuit-build-metadata.workspace = true
clap.workspace = true
console-subscriber.optional = true
@@ -211,8 +213,6 @@ opentelemetry.optional = true
opentelemetry.workspace = true
opentelemetry-otlp.optional = true
opentelemetry-otlp.workspace = true
opentelemetry-jaeger-propagator.optional = true
opentelemetry-jaeger-propagator.workspace = true
opentelemetry_sdk.optional = true
opentelemetry_sdk.workspace = true
sentry-tower.optional = true
+1 -1
View File
@@ -94,7 +94,7 @@ pub(crate) fn init(
let otlp_layer = config.allow_otlp.then(|| {
opentelemetry::global::set_text_map_propagator(
opentelemetry_jaeger_propagator::Propagator::new(),
opentelemetry_sdk::propagation::TraceContextPropagator::new(),
);
let exporter = opentelemetry_otlp::SpanExporter::builder()
+21 -1
View File
@@ -1,10 +1,12 @@
#![cfg(feature = "sentry_telemetry")]
use std::{
borrow::Cow,
str::FromStr,
sync::{Arc, OnceLock},
};
use conduwuit_build_metadata as build;
use conduwuit_core::{config::Config, debug, trace};
use sentry::{
Breadcrumb, ClientOptions, Level,
@@ -44,7 +46,7 @@ fn options(config: &Config) -> ClientOptions {
server_name,
traces_sample_rate: config.sentry_traces_sample_rate,
debug: cfg!(debug_assertions),
release: sentry::release_name!(),
release: release_name(),
user_agent: conduwuit_core::version::user_agent().into(),
attach_stacktrace: config.sentry_attach_stacktrace,
before_send: Some(Arc::new(before_send)),
@@ -91,3 +93,21 @@ fn before_breadcrumb(crumb: Breadcrumb) -> Option<Breadcrumb> {
trace!("Sentry breadcrumb: {crumb:?}");
Some(crumb)
}
fn release_name() -> Option<Cow<'static, str>> {
static RELEASE: OnceLock<Option<String>> = OnceLock::new();
RELEASE
.get_or_init(|| {
let pkg_name = env!("CARGO_PKG_NAME");
let pkg_version = env!("CARGO_PKG_VERSION");
if let Some(commit_short) = build::GIT_COMMIT_HASH_SHORT {
Some(format!("{pkg_name}@{pkg_version}+{commit_short}"))
} else {
Some(format!("{pkg_name}@{pkg_version}"))
}
})
.as_ref()
.map(|s| Cow::Borrowed(s.as_str()))
}
-1
View File
@@ -40,7 +40,6 @@ io_uring = [
"conduwuit-admin/io_uring",
"conduwuit-api/io_uring",
"conduwuit-service/io_uring",
"conduwuit-api/io_uring",
]
jemalloc = [
"conduwuit-admin/jemalloc",
+2 -2
View File
@@ -65,7 +65,7 @@ pub(crate) async fn start(server: Arc<Server>) -> Result<Arc<Services>> {
let services = Services::build(server).await?.start().await?;
#[cfg(all(feature = "systemd", target_os = "linux"))]
sd_notify::notify(true, &[sd_notify::NotifyState::Ready])
sd_notify::notify(false, &[sd_notify::NotifyState::Ready])
.expect("failed to notify systemd of ready state");
debug!("Started");
@@ -78,7 +78,7 @@ pub(crate) async fn stop(services: Arc<Services>) -> Result<()> {
debug!("Shutting down...");
#[cfg(all(feature = "systemd", target_os = "linux"))]
sd_notify::notify(true, &[sd_notify::NotifyState::Stopping])
sd_notify::notify(false, &[sd_notify::NotifyState::Stopping])
.expect("failed to notify systemd of stopping state");
// Wait for all completions before dropping or we'll lose them to the module
+8 -1
View File
@@ -67,6 +67,9 @@ release_max_log_level = [
"tracing/max_level_trace",
"tracing/release_max_level_info",
]
systemd = [
"dep:sd-notify",
]
url_preview = [
"dep:image",
"dep:webpage",
@@ -105,7 +108,7 @@ rustyline-async.workspace = true
rustyline-async.optional = true
serde_json.workspace = true
serde.workspace = true
serde_yml.workspace = true
serde-saphyr.workspace = true
sha2.workspace = true
termimad.workspace = true
termimad.optional = true
@@ -119,5 +122,9 @@ blurhash.optional = true
recaptcha-verify = { version = "0.1.5", default-features = false }
ctor.workspace = true
[target.'cfg(all(unix, target_os = "linux"))'.dependencies]
sd-notify.workspace = true
sd-notify.optional = true
[lints]
workspace = true
+3 -2
View File
@@ -129,13 +129,14 @@ pub fn changes_since<'a>(
&'a self,
room_id: Option<&'a RoomId>,
user_id: &'a UserId,
since: u64,
since: Option<u64>,
to: Option<u64>,
) -> impl Stream<Item = AnyRawAccountDataEvent> + Send + 'a {
type Key<'a> = (Option<&'a RoomId>, &'a UserId, u64, Ignore);
// Skip the data that's exactly at since, because we sent that last time
let first_possible = (room_id, user_id, since.saturating_add(1));
// ...unless this is an initial sync, in which case send everything
let first_possible = (room_id, user_id, since.map_or(0, |since| since.saturating_add(1)));
self.db
.roomuserdataid_accountdata
+1 -1
View File
@@ -271,7 +271,7 @@ pub async fn get_db_registration(&self, id: &str) -> Result<Registration> {
.id_appserviceregistrations
.get(id)
.await
.and_then(|ref bytes| serde_yml::from_slice(bytes).map_err(Into::into))
.and_then(|ref bytes| serde_saphyr::from_slice(bytes).map_err(Into::into))
.map_err(|e| err!(Database("Invalid appservice {id:?} registration: {e:?}")))
}
+6 -3
View File
@@ -45,13 +45,16 @@ fn deref(&self) -> &Self::Target { &self.server.config }
fn handle_reload(&self) -> Result {
if self.server.config.config_reload_signal {
#[cfg(all(feature = "systemd", target_os = "linux"))]
sd_notify::notify(true, &[sd_notify::NotifyState::Reloading])
.expect("failed to notify systemd of reloading state");
sd_notify::notify(false, &[
sd_notify::NotifyState::Reloading,
sd_notify::NotifyState::monotonic_usec_now().expect("Failed to read monotonic time"),
])
.expect("failed to notify systemd of reloading state");
self.reload(iter::empty())?;
#[cfg(all(feature = "systemd", target_os = "linux"))]
sd_notify::notify(true, &[sd_notify::NotifyState::Ready])
sd_notify::notify(false, &[sd_notify::NotifyState::Ready])
.expect("failed to notify systemd of ready state");
}
+29 -26
View File
@@ -11,7 +11,10 @@
use base64::{Engine as _, engine::general_purpose};
use conduwuit::{
Err, Result, Server, debug, debug_error, debug_info, debug_warn, err, error, trace,
utils::{self, MutexMap},
utils::{
self, MutexMap,
time::{self, TimeDirection},
},
warn,
};
use ruma::{Mxc, OwnedMxcUri, UserId, http_headers::ContentDisposition};
@@ -90,17 +93,22 @@ pub async fn create(
file: &[u8],
) -> Result<()> {
// Width, Height = 0 if it's not a thumbnail
let key = self.db.create_file_metadata(
mxc,
user,
&Dim::default(),
content_disposition,
content_type,
)?;
let key = self
.db
.create_file_metadata(mxc, user, &Dim::default(), content_disposition, content_type)
.map_err(|e| {
err!(Database(error!("Failed to create media metadata for MXC {mxc}: {e}")))
})?;
//TODO: Dangling metadata in database if creation fails
let mut f = self.create_media_file(&key).await?;
f.write_all(file).await?;
let mut f = self.create_media_file(&key).await.map_err(|e| {
err!(Database(error!(
"Failed to create media file for MXC {mxc} at key {key:?}: {e}"
)))
})?;
f.write_all(file).await.map_err(|e| {
err!(Database(error!("Failed to write media file for MXC {mxc} at key {key:?}: {e}")))
})?;
Ok(())
}
@@ -221,13 +229,12 @@ pub async fn get_all_mxcs(&self) -> Result<Vec<OwnedMxcUri>> {
Ok(mxcs)
}
/// Deletes all remote only media files in the given at or after
/// time/duration. Returns a usize with the amount of media files deleted.
pub async fn delete_all_remote_media_at_after_time(
/// Deletes all media files in the given time frame.
/// Returns a usize with the amount of media files deleted.
pub async fn delete_all_media_within_timeframe(
&self,
time: SystemTime,
before: bool,
after: bool,
time_boundary: SystemTime,
direction: TimeDirection,
yes_i_want_to_delete_local_media: bool,
) -> Result<usize> {
let all_keys = self.db.get_all_media_keys().await;
@@ -294,18 +301,14 @@ pub async fn delete_all_remote_media_at_after_time(
debug!("File created at: {file_created_at:?}");
if file_created_at >= time && before {
if time::is_within_bounds(file_created_at, time_boundary, direction) {
debug!(
"File is within (before) user duration, pushing to list of file paths and \
keys to delete."
);
remote_mxcs.push(mxc.to_string());
} else if file_created_at <= time && after {
debug!(
"File is not within (after) user duration, pushing to list of file paths \
and keys to delete."
"File is within bounds ({direction:?} {time_boundary:?}), pushing to list \
of file paths and keys to delete.",
);
remote_mxcs.push(mxc.to_string());
} else {
debug!("File is outside bounds ({direction:?} {time_boundary:?}), ignoring.");
}
}
@@ -313,7 +316,7 @@ pub async fn delete_all_remote_media_at_after_time(
return Err!(Database("Did not found any eligible MXCs to delete."));
}
debug_info!("Deleting media now in the past {time:?}");
debug_info!("Deleting media now {direction:?} {time_boundary:?}");
let mut deletion_count: usize = 0;
+95 -6
View File
@@ -1,7 +1,7 @@
use std::cmp;
use std::{cmp, collections::HashMap};
use conduwuit::{
Err, Result, debug, debug_info, debug_warn, error, info,
Err, Pdu, Result, debug, debug_info, debug_warn, error, info,
result::NotFound,
utils::{
IterStream, ReadyExt,
@@ -13,14 +13,16 @@
use futures::{FutureExt, StreamExt, TryStreamExt};
use itertools::Itertools;
use ruma::{
OwnedUserId, RoomId, UserId,
OwnedRoomId, OwnedUserId, RoomId, UserId,
events::{
GlobalAccountDataEventType, push_rules::PushRulesEvent, room::member::MembershipState,
GlobalAccountDataEventType, StateEventType, push_rules::PushRulesEvent,
room::member::MembershipState,
},
push::Ruleset,
serde::Raw,
};
use crate::{Services, media};
use crate::{Services, media, rooms::short::ShortStateHash};
/// The current schema version.
/// - If database is opened at greater version we reject with error. The
@@ -152,6 +154,14 @@ async fn migrate(services: &Services) -> Result<()> {
info!("Migration: Bumped database version to 18");
}
if db["global"]
.get(POPULATED_USERROOMID_LEFTSTATE_TABLE_MARKER)
.await
.is_not_found()
{
populate_userroomid_leftstate_table(services).await?;
}
assert_eq!(
services.globals.db.database_version().await,
DATABASE_VERSION,
@@ -456,7 +466,11 @@ async fn retroactively_fix_bad_data_from_roomuserid_joined(services: &Services)
for user_id in &non_joined_members {
debug_info!("User is left or banned, marking as left");
services.rooms.state_cache.mark_as_left(user_id, room_id);
services
.rooms
.state_cache
.mark_as_left(user_id, room_id, None)
.await;
}
}
@@ -624,3 +638,78 @@ async fn fix_corrupt_msc4133_fields(services: &Services) -> Result {
db.db.sort()?;
Ok(())
}
const POPULATED_USERROOMID_LEFTSTATE_TABLE_MARKER: &str = "populate_userroomid_leftstate_table";
async fn populate_userroomid_leftstate_table(services: &Services) -> Result {
type KeyVal<'a> = (Key<'a>, Raw<Option<Pdu>>);
type Key<'a> = (&'a UserId, &'a RoomId);
let db = &services.db;
let cork = db.cork_and_sync();
let userroomid_leftstate = db["userroomid_leftstate"].clone();
let (total, fixed, _) = userroomid_leftstate
.stream()
.try_fold(
(0_usize, 0_usize, HashMap::<OwnedRoomId, ShortStateHash>::new()),
async |(mut total, mut fixed, mut shortstatehash_cache): (
usize,
usize,
HashMap<_, _>,
),
((user_id, room_id), state): KeyVal<'_>|
-> Result<(usize, usize, HashMap<_, _>)> {
if state.deserialize().is_err() {
let latest_shortstatehash =
if let Some(shortstatehash) = shortstatehash_cache.get(room_id) {
*shortstatehash
} else if let Ok(shortstatehash) =
services.rooms.state.get_room_shortstatehash(room_id).await
{
shortstatehash_cache.insert(room_id.to_owned(), shortstatehash);
shortstatehash
} else {
warn!(?room_id, ?user_id, "room has no shortstatehash");
return Ok((total, fixed, shortstatehash_cache));
};
let leave_state_event = services
.rooms
.state_accessor
.state_get(
latest_shortstatehash,
&StateEventType::RoomMember,
user_id.as_str(),
)
.await;
match leave_state_event {
| Ok(leave_state_event) => {
userroomid_leftstate.put((user_id, room_id), Json(leave_state_event));
fixed = fixed.saturating_add(1);
},
| Err(_) => {
warn!(
?room_id,
?user_id,
"room cached as left has no leave event for user, removing \
cache entry"
);
userroomid_leftstate.del((user_id, room_id));
},
}
}
total = total.saturating_add(1);
Ok((total, fixed, shortstatehash_cache))
},
)
.await?;
drop(cork);
info!(?total, ?fixed, "Fixed entries in `userroomid_leftstate`.");
db["global"].insert(POPULATED_USERROOMID_LEFTSTATE_TABLE_MARKER, []);
db.db.sort()?;
Ok(())
}
+1 -3
View File
@@ -188,9 +188,7 @@ pub fn local_aliases_for_room<'a>(
}
#[tracing::instrument(skip(self), level = "debug")]
pub fn all_local_aliases<'a>(
&'a self,
) -> impl Stream<Item = (&'a RoomId, &'a str)> + Send + 'a {
pub fn all_local_aliases(&self) -> impl Stream<Item = (&RoomId, &str)> + Send + '_ {
self.db
.alias_roomid
.stream()
@@ -5,9 +5,9 @@
use std::time::Duration;
use conduwuit::{Err, Event, PduEvent, Result, debug, implement, warn};
use conduwuit::{Err, Event, PduEvent, Result, debug, debug_info, implement, trace, warn};
use ruma::{
RoomId, ServerName,
CanonicalJsonObject, RoomId, ServerName,
api::federation::room::policy::v1::Request as PolicyRequest,
events::{StateEventType, room::policy::RoomPolicyEventContent},
};
@@ -25,7 +25,25 @@
/// fail-open operation.
#[implement(super::Service)]
#[tracing::instrument(skip_all, level = "debug")]
pub async fn ask_policy_server(&self, pdu: &PduEvent, room_id: &RoomId) -> Result<bool> {
pub async fn ask_policy_server(
&self,
pdu: &PduEvent,
pdu_json: &CanonicalJsonObject,
room_id: &RoomId,
) -> Result<bool> {
if !self.services.server.config.enable_msc4284_policy_servers {
return Ok(true); // don't ever contact policy servers
}
if self.services.server.config.policy_server_check_own_events
&& pdu.origin.is_some()
&& self
.services
.server
.is_ours(pdu.origin.as_ref().unwrap().as_str())
{
return Ok(true); // don't contact policy servers for locally generated events
}
if *pdu.event_type() == StateEventType::RoomPolicy.into() {
debug!(
room_id = %room_id,
@@ -47,12 +65,12 @@ pub async fn ask_policy_server(&self, pdu: &PduEvent, room_id: &RoomId) -> Resul
let via = match policyserver.via {
| Some(ref via) => ServerName::parse(via)?,
| None => {
debug!("No policy server configured for room {room_id}");
trace!("No policy server configured for room {room_id}");
return Ok(true);
},
};
if via.is_empty() {
debug!("Policy server is empty for room {room_id}, skipping spam check");
trace!("Policy server is empty for room {room_id}, skipping spam check");
return Ok(true);
}
if !self.services.state_cache.server_in_room(via, room_id).await {
@@ -66,12 +84,12 @@ pub async fn ask_policy_server(&self, pdu: &PduEvent, room_id: &RoomId) -> Resul
let outgoing = self
.services
.sending
.convert_to_outgoing_federation_event(pdu.to_canonical_object())
.convert_to_outgoing_federation_event(pdu_json.clone())
.await;
debug!(
debug_info!(
room_id = %room_id,
via = %via,
outgoing = ?outgoing,
outgoing = ?pdu_json,
"Checking event for spam with policy server"
);
let response = tokio::time::timeout(
@@ -85,7 +103,10 @@ pub async fn ask_policy_server(&self, pdu: &PduEvent, room_id: &RoomId) -> Resul
)
.await;
let response = match response {
| Ok(Ok(response)) => response,
| Ok(Ok(response)) => {
debug!("Response from policy server: {:?}", response);
response
},
| Ok(Err(e)) => {
warn!(
via = %via,
@@ -97,16 +118,18 @@ pub async fn ask_policy_server(&self, pdu: &PduEvent, room_id: &RoomId) -> Resul
// default.
return Err(e);
},
| Err(_) => {
| Err(elapsed) => {
warn!(
via = %via,
%via,
event_id = %pdu.event_id(),
room_id = %room_id,
%room_id,
%elapsed,
"Policy server request timed out after 10 seconds"
);
return Err!("Request to policy server timed out");
},
};
trace!("Recommendation from policy server was {}", response.recommendation);
if response.recommendation == "spam" {
warn!(
via = %via,
@@ -255,7 +255,10 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu<Pdu>(
// 14-pre. If the event is not a state event, ask the policy server about it
if incoming_pdu.state_key.is_none() {
debug!(event_id = %incoming_pdu.event_id, "Checking policy server for event");
match self.ask_policy_server(&incoming_pdu, room_id).await {
match self
.ask_policy_server(&incoming_pdu, &incoming_pdu.to_canonical_object(), room_id)
.await
{
| Ok(false) => {
warn!(
event_id = %incoming_pdu.event_id,
+5 -3
View File
@@ -39,7 +39,7 @@ pub enum Status {
Seen(u64),
}
pub type Witness = HashSet<OwnedUserId>;
pub type MemberSet = HashSet<OwnedUserId>;
type Key<'a> = (&'a UserId, Option<&'a DeviceId>, &'a RoomId, &'a UserId);
impl crate::Service for Service {
@@ -67,9 +67,11 @@ pub async fn reset(&self, ctx: &Context<'_>) {
.await;
}
/// Returns only the subset of `senders` which should be sent to the client
/// according to the provided lazy loading context.
#[implement(Service)]
#[tracing::instrument(name = "retain", level = "debug", skip_all)]
pub async fn witness_retain(&self, senders: Witness, ctx: &Context<'_>) -> Witness {
pub async fn retain_lazy_members(&self, senders: MemberSet, ctx: &Context<'_>) -> MemberSet {
debug_assert!(
ctx.options.is_none_or(Options::is_enabled),
"lazy loading should be enabled by your options"
@@ -84,7 +86,7 @@ pub async fn witness_retain(&self, senders: Witness, ctx: &Context<'_>) -> Witne
pin_mut!(witness);
let _cork = self.db.db.cork();
let mut senders = Witness::with_capacity(senders.len());
let mut senders = MemberSet::with_capacity(senders.len());
while let Some((status, sender)) = witness.next().await {
if include_redundant || status == Status::Unseen {
senders.insert(sender.into());
+4 -4
View File
@@ -7,7 +7,7 @@
use database::{Deserialized, Json, Map};
use futures::{Stream, StreamExt};
use ruma::{
CanonicalJsonObject, RoomId, UserId,
CanonicalJsonObject, OwnedUserId, RoomId, UserId,
events::{AnySyncEphemeralRoomEvent, receipt::ReceiptEvent},
serde::Raw,
};
@@ -25,7 +25,7 @@ struct Services {
globals: Dep<globals::Service>,
}
pub(super) type ReceiptItem<'a> = (&'a UserId, u64, Raw<AnySyncEphemeralRoomEvent>);
pub(super) type ReceiptItem = (OwnedUserId, u64, Raw<AnySyncEphemeralRoomEvent>);
impl Data {
pub(super) fn new(args: &crate::Args<'_>) -> Self {
@@ -65,7 +65,7 @@ pub(super) fn readreceipts_since<'a>(
&'a self,
room_id: &'a RoomId,
since: u64,
) -> impl Stream<Item = ReceiptItem<'a>> + Send + 'a {
) -> impl Stream<Item = ReceiptItem> + Send + 'a {
type Key<'a> = (&'a RoomId, u64, &'a UserId);
type KeyVal<'a> = (Key<'a>, CanonicalJsonObject);
@@ -81,7 +81,7 @@ pub(super) fn readreceipts_since<'a>(
let event = serde_json::value::to_raw_value(&json)?;
Ok((user_id, count, Raw::from_json(event)))
Ok((user_id.to_owned(), count, Raw::from_json(event)))
})
.ignore_err()
}
+5 -5
View File
@@ -104,16 +104,16 @@ pub async fn private_read_get(
Ok(Raw::from_json(event))
}
/// Returns an iterator over the most recent read_receipts in a room that
/// happened after the event with id `since`.
/// Returns an iterator over the most recent read_receipts in a room,
/// optionally after the event with id `since`.
#[inline]
#[tracing::instrument(skip(self), level = "debug")]
pub fn readreceipts_since<'a>(
&'a self,
room_id: &'a RoomId,
since: u64,
) -> impl Stream<Item = ReceiptItem<'a>> + Send + 'a {
self.db.readreceipts_since(room_id, since)
since: Option<u64>,
) -> impl Stream<Item = ReceiptItem> + Send + 'a {
self.db.readreceipts_since(room_id, since.unwrap_or(0))
}
/// Sets a private read marker at PDU `count`.
+31 -3
View File
@@ -1,10 +1,18 @@
use std::{borrow::Borrow, fmt::Debug, mem::size_of_val, sync::Arc};
pub use conduwuit::matrix::pdu::{ShortEventId, ShortId, ShortRoomId, ShortStateKey};
use conduwuit::{Result, err, implement, matrix::StateKey, utils, utils::IterStream};
use conduwuit::{
Result, err, implement,
matrix::StateKey,
pair_of,
utils::{self, IterStream, ReadyExt},
};
use database::{Deserialized, Get, Map, Qry};
use futures::{Stream, StreamExt};
use ruma::{EventId, RoomId, events::StateEventType};
use futures::{
Stream, StreamExt,
stream::{self},
};
use ruma::{EventId, OwnedEventId, RoomId, events::StateEventType};
use serde::Deserialize;
use crate::{Dep, globals};
@@ -258,3 +266,23 @@ pub async fn get_or_create_shortroomid(&self, room_id: &RoomId) -> ShortRoomId {
short
})
}
#[implement(Service)]
pub async fn multi_get_state_from_short<'a, S>(
&'a self,
short_state: S,
) -> impl Stream<Item = Result<((StateEventType, StateKey), OwnedEventId)>> + Send + 'a
where
S: Stream<Item = (ShortStateKey, ShortEventId)> + Send + 'a,
{
let (short_state_keys, short_event_ids): pair_of!(Vec<_>) = short_state.unzip().await;
StreamExt::zip(
self.multi_get_statekey_from_short(stream::iter(short_state_keys.into_iter())),
self.multi_get_eventid_from_short(stream::iter(short_event_ids.into_iter())),
)
.ready_filter_map(|state_event| match state_event {
| (Ok(state_key), Ok(event_id)) => Some(Ok((state_key, event_id))),
| (Err(e), _) | (_, Err(e)) => Some(Err(e)),
})
}

Some files were not shown because too many files have changed in this diff Show More