fix(gateway): truncate reads on an expired outbox entry (#2934)

fluxer_api/src/api/push/ApnsPushService.ts

@@ -204,9 +204,7 @@ function buildApnsPayload(payload: Record<string, unknown>): Record<string, unkn
 	if (badge !== undefined) {
 		aps.badge = badge;
 	}
-	if (imageUrl) {
-		aps['mutable-content'] = 1;
-	}
+	aps['mutable-content'] = 1;
 	return {
 		...data,
 		title,

fluxer_api/src/api/push/tests/ApnsPushService.test.ts

@@ -139,7 +139,7 @@ describe('ApnsPushService', () => {
 			notification: {title: 'Alice', body: 'Hello', icon: 'https://cdn.example/avatar.png'},
 		});
 		expect(payload.image_url).toBeUndefined();
-		expect(payload.aps).not.toHaveProperty('mutable-content');
+		expect(payload.aps).toHaveProperty('mutable-content', 1);
 		expect(payload.author_avatar_url).toBe('https://cdn.example/avatar.png');
 	});
 	it('imports the APNs signing key once per PEM and rejects a truncated one every time', async () => {

fluxer_gateway/src/push/push_notification_format.erl

@@ -17,6 +17,7 @@
 
 -define(MAX_MENTIONS_FOR_PUSH, 50).
 -define(MAX_PREVIEW_BYTES, 100).
+-define(MAX_IMAGE_URL_BYTES, 1024).
 -define(CHANNEL_TYPE_GUILD_TEXT, 0).
 -define(CHANNEL_TYPE_GUILD_VOICE, 2).
 -define(CHANNEL_TYPE_GUILD_CATEGORY, 4).
@@ -487,11 +488,23 @@ format_name_list(Names) ->
 
 -spec extract_image_url(map()) -> binary() | undefined.
 extract_image_url(MessageData) ->
+    bounded_image_url(resolve_image_url(MessageData)).
+
+-spec resolve_image_url(map()) -> binary() | undefined.
+resolve_image_url(MessageData) ->
     case extract_attachment_image_url(maps:get(<<"attachments">>, MessageData, [])) of
         undefined -> extract_embed_image_url(maps:get(<<"embeds">>, MessageData, []));
         ImageUrl -> ImageUrl
     end.
 
+-spec bounded_image_url(binary() | undefined) -> binary() | undefined.
+bounded_image_url(ImageUrl) when
+    is_binary(ImageUrl), byte_size(ImageUrl) =< ?MAX_IMAGE_URL_BYTES
+->
+    ImageUrl;
+bounded_image_url(_ImageUrl) ->
+    undefined.
+
 -spec extract_attachment_image_url(term()) -> binary() | undefined.
 extract_attachment_image_url([Attachment | Rest]) when is_map(Attachment) ->
     case check_media_safe_url(Attachment) of
@@ -632,6 +645,30 @@ build_url_dm_test() ->
 build_url_guild_test() ->
     ?assertEqual(<<"/channels/123/456/789">>, build_url(123, 456, 789)).
 
+extract_image_url_keeps_a_url_within_the_size_bound_test() ->
+    Url = image_url_of_size(?MAX_IMAGE_URL_BYTES),
+    ?assertEqual(Url, extract_image_url(message_with_embed_image(Url))).
+
+extract_image_url_rejects_an_oversized_url_test() ->
+    Url = image_url_of_size(?MAX_IMAGE_URL_BYTES + 1),
+    ?assertEqual(undefined, extract_image_url(message_with_embed_image(Url))).
+
+image_url_of_size(Size) ->
+    Prefix = <<"https://media.example/">>,
+    <<Prefix/binary, (binary:copy(<<"a">>, Size - byte_size(Prefix)))/binary>>.
+
+message_with_embed_image(Url) ->
+    #{
+        <<"embeds">> => [
+            #{
+                <<"image">> => #{
+                    <<"content_type">> => <<"image/png">>,
+                    <<"proxy_url">> => Url
+                }
+            }
+        ]
+    }.
+
 extract_image_url_rejects_malformed_flags_test() ->
     MessageData = #{
         <<"attachments">> => [

fluxer_gateway/src/push/push_outbox.erl

@@ -437,10 +437,17 @@ handle_result({error, Reason}, Entry, State) ->
         attempts => maps:get(attempts, Entry)
     }),
     case is_expired(Entry, State) of
-        true -> fall_back(Entry, State);
+        true -> fall_back_prepared(Entry, State);
         false -> retry_settled(settle_if_stale(Entry, State))
     end.
 
+-spec fall_back_prepared(entry(), state()) -> state().
+fall_back_prepared(Entry, State) ->
+    case prepare(Entry, State) of
+        {skip, State1} -> State1;
+        {send, Prepared, State1} -> fall_back(Prepared, State1)
+    end.
+
 -spec retry_settled(settled()) -> state().
 retry_settled({keep, Entry, State}) ->
     schedule_retry(Entry, State);
@@ -661,3 +668,80 @@ app_pos_integer(Key, Default) ->
         Value when is_integer(Value), Value > 0 -> Value;
         _ -> Default
     end.
+
+-ifdef(TEST).
+-include_lib("eunit/include/eunit.hrl").
+
+an_expired_entry_is_handed_back_without_the_users_who_read_it_test() ->
+    Self = self(),
+    State = test_state(#{{7, 20} => {30, now_ms()}}),
+    Entry = test_entry([7, 8], fun(UserIds) -> Self ! {handed_back, UserIds} end),
+    Result = handle_result({error, timeout}, Entry, State),
+    ?assertEqual(1, count(fallbacks, Result)),
+    ?assertEqual(1, count(truncations, Result)),
+    receive
+        {handed_back, UserIds} -> ?assertEqual([8], UserIds)
+    after 2000 -> erlang:error(no_fallback_ran)
+    end.
+
+an_expired_entry_every_recipient_read_is_not_handed_back_test() ->
+    Self = self(),
+    State = test_state(#{{7, 20} => {30, now_ms()}, {8, 20} => {31, now_ms()}}),
+    Entry = test_entry([7, 8], fun(UserIds) -> Self ! {handed_back, UserIds} end),
+    Result = handle_result({error, timeout}, Entry, State),
+    ?assertEqual(0, count(fallbacks, Result)),
+    ?assertEqual(2, count(truncations, Result)),
+    receive
+        {handed_back, _} -> erlang:error(fallback_ran_for_read_users)
+    after 200 -> ok
+    end.
+
+an_expired_clear_is_handed_back_unchanged_test() ->
+    Self = self(),
+    State = test_state(#{{7, 20} => {30, now_ms()}, {8, 20} => {30, now_ms()}}),
+    Entry = (test_entry([7, 8], fun(UserIds) -> Self ! {handed_back, UserIds} end))#{
+        kind := clear
+    },
+    Result = handle_result({error, timeout}, Entry, State),
+    ?assertEqual(1, count(fallbacks, Result)),
+    receive
+        {handed_back, UserIds} -> ?assertEqual([7, 8], UserIds)
+    after 2000 -> erlang:error(no_fallback_ran)
+    end.
+
+test_state(Reads) ->
+    #{
+        jobs => gb_trees:empty(),
+        ready => queue:new(),
+        inflight => #{},
+        fallback_backlog => queue:new(),
+        fallback_runners => #{},
+        next_seq => 1,
+        reads => Reads,
+        active => #{},
+        counters => #{},
+        max_queue => ?DEFAULT_MAX_QUEUE,
+        max_inflight => ?DEFAULT_MAX_INFLIGHT,
+        max_fallback_runners => ?DEFAULT_MAX_FALLBACK_RUNNERS,
+        request_timeout_ms => ?DEFAULT_REQUEST_TIMEOUT_MS,
+        max_age_ms => ?DEFAULT_MAX_AGE_MS,
+        retry_base_ms => ?DEFAULT_RETRY_BASE_MS
+    }.
+
+test_entry(UserIds, Fallback) ->
+    #{
+        kind => message,
+        subject => <<"rpc.push.message">>,
+        job => #{<<"user_ids">> => [integer_to_binary(UserId) || UserId <- UserIds]},
+        body => <<"{}">>,
+        user_ids => UserIds,
+        channel_id => 20,
+        message_id => 30,
+        fallback => Fallback,
+        seq => 0,
+        enqueued_at => now_ms() - ?DEFAULT_MAX_AGE_MS,
+        attempts => 3,
+        config_version => undefined
+    }.
+
+-endif.

fluxer_push/src/delivery.rs

@@ -279,14 +279,11 @@ async fn run_subject<T: Transport>(
             let job_transport = transport.clone();
             running.spawn(async move {
                 let _permit = permit;
-                let mut result = run_job(&job_state, &job_sends, job).await;
+                let result = run_job(&job_state, &job_sends, job).await;
                 if matches!(result, Answer::Done) {
                     for claim in claims {
                         claim.done();
                     }
-                    if recipients_running {
-                        result = Answer::NotDone(RUNNING);
-                    }
                 }
                 answer(&job_transport, reply_to, result).await;
             });

fluxer_push/src/payload.rs

@@ -11,7 +11,8 @@ const CLEAR_ACTION: &str = "clear_channel";
 const RING_TYPE: &str = "call_ring";
 const FALLBACK_TAG: &str = "fluxer-message";
 const FALLBACK_TITLE: &str = "Fluxer";
-const APNS_CATEGORY: &str = "FLUXER_MESSAGE";
+const APNS_CATEGORY: &str = "fluxer_message";
+const FCM_CLICK_ACTION: &str = "FLUXER_MESSAGE";
 const APNS_SOUND: &str = "default";
 const APNS_ALERT_EXPIRATION_SECONDS: i64 = 86_400;
 const APNS_BACKGROUND_EXPIRATION_SECONDS: i64 = 86_400;
@@ -25,7 +26,9 @@ const RING_CALLER_NAME_KEY: &str = "caller_name";
 const RING_CALLER_AVATAR_KEY: &str = "caller_avatar_url";
 const RING_AVATAR_KEYS: [&str; 1] = [RING_CALLER_AVATAR_KEY];
 const RING_IDENTITY_KEYS: [&str; 2] = [RING_CALLER_ID_KEY, RING_CALLER_NAME_KEY];
-const MINIMAL_DATA_KEYS: [&str; 7] = [
+const MINIMAL_DATA_KEYS: [&str; 9] = [
+    "type",
+    "action",
     "channel_id",
     "message_id",
     "guild_id",
@@ -201,7 +204,7 @@ fn fcm_notification_message(device_token: &str, envelope: &Value) -> Value {
     let mut android_notification = json!({
         "channel_id": "fluxer_default_push",
         "tag": tag,
-        "click_action": APNS_CATEGORY,
+        "click_action": FCM_CLICK_ACTION,
     });
     put_image(&mut android_notification, image_url);
     json!({
@@ -262,9 +265,7 @@ pub fn apns_payload(envelope: &Value) -> Value {
     if let Some(badge) = badge_number(data.and_then(|data| data.get("badge_count"))) {
         aps.insert("badge".to_owned(), badge.into());
     }
-    if image_url.is_some() {
-        aps.insert("mutable-content".to_owned(), 1.into());
-    }
+    aps.insert("mutable-content".to_owned(), 1.into());
     payload.insert("title".to_owned(), title.into());
     payload.insert("body".to_owned(), body.into());
     payload.remove("url");
@@ -591,3 +592,95 @@ fn non_empty(value: &Value) -> Option<&str> {
 fn serialize(value: &Value) -> Vec<u8> {
     serde_json::to_vec(value).expect("a json value serialises")
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::job::{ClearJob, MessageJob, NotificationFields};
+
+    const CHANNEL_ID: &str = "9876543210987654321";
+    const MESSAGE_ID: &str = "1122334455667788990";
+    const USER_ID: &str = "1234567890123456789";
+
+    fn message_job(image_url: Option<&str>) -> MessageJob {
+        MessageJob {
+            v: 1,
+            config_version: 7,
+            guild_id: "0".to_owned(),
+            channel_id: CHANNEL_ID.to_owned(),
+            message_id: MESSAGE_ID.to_owned(),
+            notification: NotificationFields {
+                title: "Elias".to_owned(),
+                body: "see you tomorrow".to_owned(),
+                icon: "https://media.fluxer.app/avatars/1/a.webp".to_owned(),
+                badge: "https://media.fluxer.app/badge.png".to_owned(),
+                tag: format!("channel:{CHANNEL_ID}:{MESSAGE_ID}"),
+                notification_tag: format!("channel:{CHANNEL_ID}"),
+                url: format!("/channels/@me/{CHANNEL_ID}"),
+                image_url: image_url.map(str::to_owned),
+            },
+            user_ids: vec![USER_ID.to_owned()],
+        }
+    }
+
+    fn clear_job() -> ClearJob {
+        ClearJob {
+            v: 1,
+            config_version: 7,
+            user_id: USER_ID.to_owned(),
+            channel_id: CHANNEL_ID.to_owned(),
+            message_id: MESSAGE_ID.to_owned(),
+        }
+    }
+
+    fn apns_for(image_url: Option<&str>) -> Value {
+        apns_payload(&web_push_message(&message_job(image_url), USER_ID, 3))
+    }
+
+    #[test]
+    fn a_message_without_an_image_still_runs_the_notification_service_extension() {
+        assert_eq!(apns_for(None)["aps"]["mutable-content"], json!(1));
+    }
+
+    #[test]
+    fn a_message_with_an_image_still_runs_the_notification_service_extension() {
+        assert_eq!(
+            apns_for(Some("https://media.fluxer.app/proxy/attachment.webp"))["aps"]["mutable-content"],
+            json!(1)
+        );
+    }
+
+    #[test]
+    fn a_background_clear_never_runs_the_notification_service_extension() {
+        let payload = apns_payload(&web_push_clear(&clear_job(), 3));
+        assert_eq!(payload["aps"], json!({"content-available": 1}));
+    }
+
+    #[test]
+    fn the_apns_category_is_the_identifier_the_app_registers() {
+        assert_eq!(apns_for(None)["aps"]["category"], json!("fluxer_message"));
+    }
+
+    #[test]
+    fn the_android_click_action_keeps_its_own_identifier() {
+        let message = fcm_message(
+            "device-token",
+            &web_push_message(&message_job(None), USER_ID, 3),
+        );
+        assert_eq!(
+            message["message"]["android"]["notification"]["click_action"],
+            json!("FLUXER_MESSAGE")
+        );
+    }
+
+    #[test]
+    fn a_shrunk_clear_still_tells_the_client_it_is_a_clear() {
+        let envelope = web_push_clear(&clear_job(), 3);
+        let budget = serialize(&envelope).len() - 1;
+        let (bytes, step) = fit(&envelope, budget);
+        assert_eq!(step, Some(PayloadShrink::Minimal));
+        let shrunk: Value = serde_json::from_slice(&bytes).expect("the shrunk payload is json");
+        assert_eq!(shrunk["data"]["type"], json!(CLEAR_TYPE));
+        assert_eq!(shrunk["data"]["action"], json!(CLEAR_ACTION));
+    }
+}

fluxer_push/src/providers/mod.rs

@@ -91,15 +91,12 @@ pub async fn send(state: &AppState, sub: &Subscription, envelope: &Value) -> Sen
         return SendOutcome::permanent("unsupported_platform");
     };
     let started_ms = now_ms();
-    let direct = own_relay::parse(&sub.endpoint, &state.cfg.own_relay_hosts);
+    let direct = in_process_hop(&sub.endpoint, &state.cfg.own_relay_hosts);
     let outcome = match (route, direct) {
         (Route::WebPush, Some(hop)) => {
             let hopped = hop.as_subscription(sub);
             state.metrics.record_own_relay_shortcut();
-            match hop.leg {
-                own_relay::Leg::Fcm => fcm::send(state, &hopped, envelope).await,
-                _ => apns::send(state, &hopped, envelope).await,
-            }
+            apns::send(state, &hopped, envelope).await
         }
         (Route::WebPush, None) => web_push::send(state, sub, envelope).await,
         (Route::LegacyApns, _) => apns::send(state, sub, envelope).await,
@@ -116,6 +113,10 @@ pub async fn send(state: &AppState, sub: &Subscription, envelope: &Value) -> Sen
     outcome
 }
 
+fn in_process_hop(endpoint: &str, hosts: &[String]) -> Option<own_relay::Hop> {
+    own_relay::parse(endpoint, hosts).filter(|hop| matches!(hop.leg, own_relay::Leg::Apns))
+}
+
 fn route_label(route: Route) -> DeliveryRoute {
     match route {
         Route::WebPush => DeliveryRoute::WebPush,
@@ -142,3 +143,32 @@ fn result_of(outcome: &SendOutcome) -> SendResult {
         SendOutcome::Transient { .. } => SendResult::Transient,
     }
 }
+
+#[cfg(test)]
+mod hop_tests {
+    use super::*;
+
+    const TOKEN: &str = "3dbc5a5ef1a1c1666afc26f466e1b3ebaaf4c66d92dddeb0fd1b69c49641d4cd";
+
+    fn ours() -> Vec<String> {
+        vec!["push.fluxer.com".to_owned()]
+    }
+
+    #[test]
+    fn only_the_plain_apns_leg_is_delivered_in_process() {
+        let apns = format!("https://push.fluxer.com/relay/v1/apns/canary/production/{TOKEN}");
+        assert!(in_process_hop(&apns, &ours()).is_some());
+    }
+
+    #[test]
+    fn an_fcm_relay_endpoint_keeps_its_encrypted_network_hop() {
+        let fcm = "https://push.fluxer.com/relay/v1/fcm/canary/tok%3AAPA91bExample";
+        assert!(in_process_hop(fcm, &ours()).is_none());
+    }
+
+    #[test]
+    fn a_pushkit_relay_endpoint_keeps_its_voip_topic() {
+        let voip = format!("https://push.fluxer.com/relay/v1/apns-voip/canary/production/{TOKEN}");
+        assert!(in_process_hop(&voip, &ours()).is_none());
+    }
+}

fluxer_push/src/providers/own_relay.rs

@@ -140,6 +140,27 @@ mod tests {
         assert_eq!(hop.device_token, "dYC_x9gXTjyyrG8_Aw3nUM:APA91bExample");
     }
 
+    #[test]
+    fn a_voip_endpoint_on_our_own_relay_keeps_its_own_leg() {
+        let hop = parse(
+            &format!("https://push.fluxer.com/relay/v1/apns-voip/canary/production/{TOKEN}"),
+            &ours(),
+        )
+        .expect("a voip endpoint parses");
+        assert_eq!(hop.leg, Leg::ApnsVoip);
+    }
+
+    #[test]
+    fn only_the_apns_alert_leg_is_taken_in_process() {
+        let shortcut = |path: &str| {
+            parse(&format!("https://push.fluxer.com/relay/v1/{path}"), &ours())
+                .filter(|hop| matches!(hop.leg, Leg::Apns))
+        };
+        assert!(shortcut(&format!("apns/canary/production/{TOKEN}")).is_some());
+        assert!(shortcut(&format!("apns-voip/canary/production/{TOKEN}")).is_none());
+        assert!(shortcut("fcm/canary/dYC_x9gXTjyyrG8_Aw3nUM%3AAPA91bExample").is_none());
+    }
+
     #[test]
     fn a_relay_we_do_not_operate_is_left_on_the_network_path() {
         let endpoint = format!("https://push.example.org/relay/v1/apns/canary/production/{TOKEN}");

fluxer_push/src/providers/web_push.rs

@@ -37,6 +37,7 @@ const OCTET_STREAM: &str = "application/octet-stream";
 const AES128GCM: &str = "aes128gcm";
 const NOT_FOUND: u16 = 404;
 const GONE: u16 = 410;
+const INSUFFICIENT_STORAGE: u16 = 507;
 const MAX_HOSTNAME_BYTES: usize = 253;
 const MAX_LABEL_BYTES: usize = 63;
 
@@ -110,7 +111,7 @@ pub async fn send(state: &AppState, sub: &Subscription, envelope: &Value) -> Sen
                 continue;
             }
         };
-        if is_transient_status(status) && attempt < MAX_TRANSIENT_RETRIES {
+        if should_retry(status, attempt) {
             tokio::time::sleep(retry_delay(attempt)).await;
             attempt += 1;
             continue;
@@ -127,6 +128,10 @@ fn delivery_headers(envelope: &Value) -> (&'static str, &'static str) {
     }
 }
 
+fn should_retry(status: u16, attempt: u32) -> bool {
+    is_transient_status(status) && status != INSUFFICIENT_STORAGE && attempt < MAX_TRANSIENT_RETRIES
+}
+
 fn classify(status: u16) -> SendOutcome {
     match status {
         200..=299 => SendOutcome::Accepted,
@@ -134,6 +139,7 @@ fn classify(status: u16) -> SendOutcome {
         NOT_FOUND => SendOutcome::TokenInvalid {
             reason: "not_found",
         },
+        INSUFFICIENT_STORAGE => SendOutcome::permanent("http_507"),
         _ if is_transient_status(status) => SendOutcome::transient(format!("http_{status}")),
         _ => SendOutcome::permanent(format!("http_{status}")),
     }
@@ -208,3 +214,43 @@ fn retry_delay(attempt: u32) -> Duration {
     let jitter = rand::rng().random_range(1..=(base / 4).max(1));
     Duration::from_millis(MAX_RETRY_DELAY_MS.min(base + jitter - 1))
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn a_unified_push_topic_with_no_listener_is_permanent() {
+        assert_eq!(
+            classify(INSUFFICIENT_STORAGE),
+            SendOutcome::permanent("http_507")
+        );
+    }
+
+    #[test]
+    fn an_unavailable_push_service_stays_retryable() {
+        assert_eq!(classify(503), SendOutcome::transient("http_503"));
+    }
+}
+
+#[cfg(test)]
+mod retry_tests {
+    use super::*;
+
+    #[test]
+    fn a_topic_with_no_listener_is_not_retried() {
+        assert!(!should_retry(INSUFFICIENT_STORAGE, 0));
+    }
+
+    #[test]
+    fn an_unavailable_push_service_is_retried_until_the_budget_runs_out() {
+        assert!(should_retry(503, 0));
+        assert!(!should_retry(503, MAX_TRANSIENT_RETRIES));
+    }
+
+    #[test]
+    fn a_permanent_status_is_never_retried() {
+        assert!(!should_retry(400, 0));
+        assert!(!should_retry(410, 0));
+    }
+}

fluxer_push/src/vendor.rs

@@ -221,7 +221,6 @@ fn apns_dead_token(status: u16, reason: &str) -> Option<DeadToken> {
         (_, "Unregistered") => Some(DeadToken::Gone("unregistered")),
         (410, _) => Some(DeadToken::Gone("gone")),
         (400, "BadDeviceToken") => Some(DeadToken::Invalid("bad_device_token")),
-        (400, "DeviceTokenNotForTopic") => Some(DeadToken::Invalid("device_token_not_for_topic")),
         _ => None,
     }
 }
@@ -308,6 +307,19 @@ mod tests {
         assert!(!error.without_url().to_string().contains(TOKEN));
     }
 
+    #[test]
+    fn a_wrong_topic_is_not_a_dead_token() {
+        assert_eq!(apns_dead_token(400, "DeviceTokenNotForTopic"), None);
+    }
+
+    #[test]
+    fn a_bad_device_token_is_still_a_dead_token() {
+        assert_eq!(
+            apns_dead_token(400, "BadDeviceToken"),
+            Some(DeadToken::Invalid("bad_device_token"))
+        );
+    }
+
     #[tokio::test]
     async fn a_refused_connection_stays_retryable() {
         let error = error_for("http://127.0.0.1:1/").await;