From 395893d3faf4a0e297274b56c0db95dd54d5ca1c Mon Sep 17 00:00:00 2001 From: Andrew Plaza Date: Thu, 2 Jul 2026 10:52:24 -0400 Subject: [PATCH] feat(proto): add KpRotation + KpDeletion task variants + regen xmtp_proto Empty payloads give each recurring singleton a stable data_hash for PullInDeadline targeting. Minimal arms keep xmtp_mls compiling standalone; real handlers land with the KP-consumer implementation. gen/ remains regenerated AHEAD of the upstream proto pin (xmtp/proto#339 carries all three messages); proto_version gets bumped to its merge SHA. --- crates/xmtp_mls/src/worker/tasks.rs | 10 + .../xmtp_proto/src/gen/proto_descriptor.bin | Bin 385724 -> 386348 bytes .../xmtp_proto/src/gen/xmtp.mls.database.rs | 34 +++- .../src/gen/xmtp.mls.database.serde.rs | 172 ++++++++++++++++++ 4 files changed, 215 insertions(+), 1 deletion(-) diff --git a/crates/xmtp_mls/src/worker/tasks.rs b/crates/xmtp_mls/src/worker/tasks.rs index d3465a8a06..a5293050e4 100644 --- a/crates/xmtp_mls/src/worker/tasks.rs +++ b/crates/xmtp_mls/src/worker/tasks.rs @@ -392,6 +392,16 @@ where .db() .pull_in_task_deadline(&p.target_data_hash, p.not_later_than_ns)?; } + Some(xmtp_proto::xmtp::mls::database::task::Task::KpRotation(_)) + | Some(xmtp_proto::xmtp::mls::database::task::Task::KpDeletion(_)) => { + // Minimal arms: real handlers land with the KP-consumer impl. + // Nothing seeds these singletons yet, so dropping is safe. + tracing::warn!( + "KP task {} received before handler landed; dropping", + task.id + ); + context.db().delete_task(task.id)?; + } None => { tracing::error!("Task {} has no data. Deleting.", task.id); context.db().delete_task(task.id)?; diff --git a/crates/xmtp_proto/src/gen/proto_descriptor.bin b/crates/xmtp_proto/src/gen/proto_descriptor.bin index 1ff059c6323059a7a562e34c1933f1ba3feef313..7404f40ddd8f236320eed551bcc02e6b49942daf 100644 GIT binary patch delta 1161 zcmbW0y>HV%7{-0)b8#JpNLg4}SP(mZ0Q?DTyyw_~g@q;3bNBw9dtSe9K7G!=`;zaUqjP+& zek7AF{+W+*E;+$Z$l#ev&QPA+q?wM_^4-t~E!Qb%991({E79{!GJR#o+fB~5m0j8y zsRS~GEn;ZwFQ~BNZ6@bKC{>YmOuSN-(m$>Xu}TXD;W1f5<8LSF7$aC&k3RV1W$7ZK zl&VBQOwH(7nx%v&y23~nPe<>+qf+!JAm4f!d5j5DbL2AAXi|(xHIXXdbCbnXsh}#C z=n&JG%0`$57#m?4U~GhGfHA=Oh+={mDJT_G4EY^5+6&3qL_wdWn-BzHg{j5pM@Vkf zi}*tPCY*s)Q7@zltcs;F-Hz+wsvO-Zg#~g-#YR_NuS1!*KhD3H164gn?`7~+6K5ExIXu~w!LCG z!72<2CWWDyd7aVMGBh->2%%s6BXqn$f(Kns_<=Bm$!&Mv=-D#$N}Bq}2!(GMHpF)W zK_Mr!ELuj`asm!1AYcISf5ufI?jppckd`pAq%5Oe+y@L;UDEUMj=(xHwTf5cxqyHj zsp9!mL1A^FmMUnuy1bHLa8<*@t%DB(i&jha0hGMcXA(?~x>lHru^z!N?D~k4dtJvg zsl?z?&*V}CE_If=U~s7?E<}n39xfaRz#3RjFxhoud^%OKayCzYVfY9dBVY5QcYWCvmeEG+D{rC6`<->2gs)!7gAUB3Awi5u5lG(KHDX5V5p4e}rHo zTw`lvXDRq2QVHpty^oa7&HFw(H?LpL$EP!$@fpm-JFX33p`+>$Z{c3VPeUB@=t)zrLFcU%YkX(Mt>tb2whmAe!+Yrn=9R%1d-QNhe@7Bjp*bCkDhn8E0cPb;go4K zTeJ%xXl$`=x))HqB^_l^7tC8eY$X+fqM=Ss6!g2joJi+yY-TkYwV{!Xc)?J*DNKNy zp{(UY>b2c^)&nhqWe?Dh%7Q?az^dj%l0fGYAW5*45Og3t1%RMh2V_, } /// Nested message and enum types in `Task`. @@ -798,6 +798,10 @@ pub mod task { ProcessPendingSelfRemove(super::ProcessPendingSelfRemove), #[prost(message, tag = "4")] PullInDeadline(super::PullInDeadline), + #[prost(message, tag = "5")] + KpRotation(super::KpRotation), + #[prost(message, tag = "6")] + KpDeletion(super::KpDeletion), } } impl ::prost::Name for Task { @@ -831,6 +835,34 @@ impl ::prost::Name for PullInDeadline { "/xmtp.mls.database.PullInDeadline".into() } } +/// Recurring singleton: rotate + upload a fresh key package when the identity's +/// rotation deadline is due. Empty payload => stable data_hash for pull-ins. +#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)] +pub struct KpRotation {} +impl ::prost::Name for KpRotation { + const NAME: &'static str = "KpRotation"; + const PACKAGE: &'static str = "xmtp.mls.database"; + fn full_name() -> ::prost::alloc::string::String { + "xmtp.mls.database.KpRotation".into() + } + fn type_url() -> ::prost::alloc::string::String { + "/xmtp.mls.database.KpRotation".into() + } +} +/// Recurring singleton: delete superseded local key-package material whose +/// delete_at_ns has passed. Empty payload => stable data_hash for pull-ins. +#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)] +pub struct KpDeletion {} +impl ::prost::Name for KpDeletion { + const NAME: &'static str = "KpDeletion"; + const PACKAGE: &'static str = "xmtp.mls.database"; + fn full_name() -> ::prost::alloc::string::String { + "xmtp.mls.database.KpDeletion".into() + } + fn type_url() -> ::prost::alloc::string::String { + "/xmtp.mls.database.KpDeletion".into() + } +} #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct SendSyncArchive { #[prost(message, optional, tag = "1")] diff --git a/crates/xmtp_proto/src/gen/xmtp.mls.database.serde.rs b/crates/xmtp_proto/src/gen/xmtp.mls.database.serde.rs index fef20891e0..1e3bb9cd79 100644 --- a/crates/xmtp_proto/src/gen/xmtp.mls.database.serde.rs +++ b/crates/xmtp_proto/src/gen/xmtp.mls.database.serde.rs @@ -1002,6 +1002,150 @@ impl<'de> serde::Deserialize<'de> for InstallationIds { deserializer.deserialize_struct("xmtp.mls.database.InstallationIds", FIELDS, GeneratedVisitor) } } +impl serde::Serialize for KpDeletion { + #[allow(deprecated)] + fn serialize(&self, serializer: S) -> std::result::Result + where + S: serde::Serializer, + { + use serde::ser::SerializeStruct; + let len = 0; + let struct_ser = serializer.serialize_struct("xmtp.mls.database.KpDeletion", len)?; + struct_ser.end() + } +} +impl<'de> serde::Deserialize<'de> for KpDeletion { + #[allow(deprecated)] + fn deserialize(deserializer: D) -> std::result::Result + where + D: serde::Deserializer<'de>, + { + const FIELDS: &[&str] = &[ + ]; + + #[allow(clippy::enum_variant_names)] + enum GeneratedField { + __SkipField__, + } + impl<'de> serde::Deserialize<'de> for GeneratedField { + fn deserialize(deserializer: D) -> std::result::Result + where + D: serde::Deserializer<'de>, + { + struct GeneratedVisitor; + + impl serde::de::Visitor<'_> for GeneratedVisitor { + type Value = GeneratedField; + + fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(formatter, "expected one of: {:?}", &FIELDS) + } + + #[allow(unused_variables)] + fn visit_str(self, value: &str) -> std::result::Result + where + E: serde::de::Error, + { + Ok(GeneratedField::__SkipField__) + } + } + deserializer.deserialize_identifier(GeneratedVisitor) + } + } + struct GeneratedVisitor; + impl<'de> serde::de::Visitor<'de> for GeneratedVisitor { + type Value = KpDeletion; + + fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str("struct xmtp.mls.database.KpDeletion") + } + + fn visit_map(self, mut map_: V) -> std::result::Result + where + V: serde::de::MapAccess<'de>, + { + while map_.next_key::()?.is_some() { + let _ = map_.next_value::()?; + } + Ok(KpDeletion { + }) + } + } + deserializer.deserialize_struct("xmtp.mls.database.KpDeletion", FIELDS, GeneratedVisitor) + } +} +impl serde::Serialize for KpRotation { + #[allow(deprecated)] + fn serialize(&self, serializer: S) -> std::result::Result + where + S: serde::Serializer, + { + use serde::ser::SerializeStruct; + let len = 0; + let struct_ser = serializer.serialize_struct("xmtp.mls.database.KpRotation", len)?; + struct_ser.end() + } +} +impl<'de> serde::Deserialize<'de> for KpRotation { + #[allow(deprecated)] + fn deserialize(deserializer: D) -> std::result::Result + where + D: serde::Deserializer<'de>, + { + const FIELDS: &[&str] = &[ + ]; + + #[allow(clippy::enum_variant_names)] + enum GeneratedField { + __SkipField__, + } + impl<'de> serde::Deserialize<'de> for GeneratedField { + fn deserialize(deserializer: D) -> std::result::Result + where + D: serde::Deserializer<'de>, + { + struct GeneratedVisitor; + + impl serde::de::Visitor<'_> for GeneratedVisitor { + type Value = GeneratedField; + + fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(formatter, "expected one of: {:?}", &FIELDS) + } + + #[allow(unused_variables)] + fn visit_str(self, value: &str) -> std::result::Result + where + E: serde::de::Error, + { + Ok(GeneratedField::__SkipField__) + } + } + deserializer.deserialize_identifier(GeneratedVisitor) + } + } + struct GeneratedVisitor; + impl<'de> serde::de::Visitor<'de> for GeneratedVisitor { + type Value = KpRotation; + + fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str("struct xmtp.mls.database.KpRotation") + } + + fn visit_map(self, mut map_: V) -> std::result::Result + where + V: serde::de::MapAccess<'de>, + { + while map_.next_key::()?.is_some() { + let _ = map_.next_value::()?; + } + Ok(KpRotation { + }) + } + } + deserializer.deserialize_struct("xmtp.mls.database.KpRotation", FIELDS, GeneratedVisitor) + } +} impl serde::Serialize for PermissionPolicyOption { #[allow(deprecated)] fn serialize(&self, serializer: S) -> std::result::Result @@ -2958,6 +3102,12 @@ impl serde::Serialize for Task { task::Task::PullInDeadline(v) => { struct_ser.serialize_field("pull_in_deadline", v)?; } + task::Task::KpRotation(v) => { + struct_ser.serialize_field("kp_rotation", v)?; + } + task::Task::KpDeletion(v) => { + struct_ser.serialize_field("kp_deletion", v)?; + } } } struct_ser.end() @@ -2978,6 +3128,10 @@ impl<'de> serde::Deserialize<'de> for Task { "processPendingSelfRemove", "pull_in_deadline", "pullInDeadline", + "kp_rotation", + "kpRotation", + "kp_deletion", + "kpDeletion", ]; #[allow(clippy::enum_variant_names)] @@ -2986,6 +3140,8 @@ impl<'de> serde::Deserialize<'de> for Task { SendSyncArchive, ProcessPendingSelfRemove, PullInDeadline, + KpRotation, + KpDeletion, __SkipField__, } impl<'de> serde::Deserialize<'de> for GeneratedField { @@ -3012,6 +3168,8 @@ impl<'de> serde::Deserialize<'de> for Task { "sendSyncArchive" | "send_sync_archive" => Ok(GeneratedField::SendSyncArchive), "processPendingSelfRemove" | "process_pending_self_remove" => Ok(GeneratedField::ProcessPendingSelfRemove), "pullInDeadline" | "pull_in_deadline" => Ok(GeneratedField::PullInDeadline), + "kpRotation" | "kp_rotation" => Ok(GeneratedField::KpRotation), + "kpDeletion" | "kp_deletion" => Ok(GeneratedField::KpDeletion), _ => Ok(GeneratedField::__SkipField__), } } @@ -3060,6 +3218,20 @@ impl<'de> serde::Deserialize<'de> for Task { return Err(serde::de::Error::duplicate_field("pullInDeadline")); } task__ = map_.next_value::<::std::option::Option<_>>()?.map(task::Task::PullInDeadline) +; + } + GeneratedField::KpRotation => { + if task__.is_some() { + return Err(serde::de::Error::duplicate_field("kpRotation")); + } + task__ = map_.next_value::<::std::option::Option<_>>()?.map(task::Task::KpRotation) +; + } + GeneratedField::KpDeletion => { + if task__.is_some() { + return Err(serde::de::Error::duplicate_field("kpDeletion")); + } + task__ = map_.next_value::<::std::option::Option<_>>()?.map(task::Task::KpDeletion) ; } GeneratedField::__SkipField__ => {