From a90991612c521e5161831796c348e4a8fa9910a3 Mon Sep 17 00:00:00 2001 From: Hampus Date: Thu, 24 Sep 2026 15:04:24 +0200 Subject: [PATCH] fix(gateway): truncate reads on an expired outbox entry (#2934) --- .../src/push/push_notification_format.erl | 37 ++++++++ fluxer_gateway/src/push/push_outbox.erl | 86 ++++++++++++++++++- 2 files changed, 122 insertions(+), 1 deletion(-) diff --git a/fluxer_gateway/src/push/push_notification_format.erl b/fluxer_gateway/src/push/push_notification_format.erl index a584075d4..8203d2129 100644 --- a/fluxer_gateway/src/push/push_notification_format.erl +++ b/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/">>, + <>, 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">> => [ diff --git a/fluxer_gateway/src/push/push_outbox.erl b/fluxer_gateway/src/push/push_outbox.erl index aaa8d7732..3d4627089 100644 --- a/fluxer_gateway/src/push/push_outbox.erl +++ b/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.