Skip to content

Skip SSE config updated events if they arrive too fast #348

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
hansott wants to merge 1 commit into
base: main
Choose a base branch
from
+34 −3
Open

Skip SSE config updated events if they arrive too fast #348

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 20 additions & 1 deletion agent_api/src/main/java/dev/aikido/agent_api/background/RealtimeSSETask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,12 +12,14 @@
import dev.aikido.agent_api.storage.ServiceConfigStore;

import java.util.Optional;
import java.util.concurrent.TimeUnit;

public class RealtimeSSETask extends Thread {
private static final Logger logger = LogManager.getLogger(RealtimeSSETask.class);
private final RealtimeSSEAPI realtimeSSEApi;
private final ReportingApiHTTP reportingApi;
private Optional<Long> configLastUpdatedAt;
private Optional<Long> lastConfigRefreshStartedAt = Optional.empty();

public RealtimeSSETask(RealtimeSSEAPI realtimeSSEApi, ReportingApiHTTP reportingApi) {
super("RealtimeSSETask");
Expand All @@ -32,7 +34,7 @@ public void run() {
realtimeSSEApi.listen(this::onEvent);
}

public void onEvent(SSEParser.Event event) {
public synchronized void onEvent(SSEParser.Event event) {
logger.trace("SSE event received: %s", event.event());
if (!"config-updated".equals(event.event())) {
return;
Expand All @@ -51,6 +53,11 @@ public void onEvent(SSEParser.Event event) {
return;
}

if (configUpdateArrivedTooFast()) {
logger.debug("Ignoring SSE config-updated event during refresh throttle");
return;
Comment on lines +56 to +58

@aikido-pr-checks aikido-pr-checks Bot Aug 19, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Medium - SSE throttle can suppress newer security config updates for up to a minute

After the first refresh starts, any later config-updated event inside the 9-second window is dropped before its configUpdatedAt value is recorded, even when it announces a newer config or the previous fetch failed. Because the only remaining recovery path is the background RealtimeTask poll that runs every 60 seconds, firewall lists, route allowlists, and outbound blocking rules can stay stale for nearly a minute and requests that should have been blocked can continue to pass during that window.

Show fix

Do not permanently discard newer SSE updates during the throttle window. Instead, coalesce them by remembering the highest pending configUpdatedAt seen while a refresh is in progress/throttled and trigger one follow-up fetch when the window expires, or only rate-limit duplicate timestamps while still processing newer versions immediately.

More info - Reply on this comment to give feedback or ignore the issue.

}

Optional<APIResponse> newConfig = reportingApi.fetchNewConfig();
if (newConfig.isEmpty()) {
logger.debug("Failed to fetch config after SSE event");
Expand All @@ -66,4 +73,16 @@ public void onEvent(SSEParser.Event event) {

logger.debug("Config updated via SSE");
}

private boolean configUpdateArrivedTooFast() {
long now = System.nanoTime();
if (lastConfigRefreshStartedAt.isPresent()) {
if (now - lastConfigRefreshStartedAt.get() < TimeUnit.SECONDS.toNanos(9)) {
return true;
}
}

lastConfigRefreshStartedAt = Optional.of(now);
return false;
}
}
16 changes: 14 additions & 2 deletions agent_api/src/test/java/background/RealtimeSSETaskTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,7 +84,7 @@ public void testHandlesFetchFailureGracefully() {
}

@Test
public void testRepeatedEventAfterFetchFailure() {
public void testThrottlesRepeatedEventAfterFetchFailure() {
when(reportingApi.fetchNewConfig())
.thenReturn(Optional.empty())
.thenReturn(Optional.of(sampleConfig(200)));
Expand All @@ -94,7 +94,19 @@ public void testRepeatedEventAfterFetchFailure() {
task.onEvent(event);
task.onEvent(event);

verify(reportingApi, times(2)).fetchNewConfig();
verify(reportingApi, times(1)).fetchNewConfig();
verify(reportingApi, never()).fetchBlockedLists();
}

@Test
public void testThrottlesNewerEvents() {
when(reportingApi.fetchNewConfig()).thenReturn(Optional.of(sampleConfig(200)));
when(reportingApi.fetchBlockedLists()).thenReturn(Optional.empty());

task.onEvent(new SSEParser.Event("config-updated", "{\"configUpdatedAt\":200}"));
task.onEvent(new SSEParser.Event("config-updated", "{\"configUpdatedAt\":300}"));

verify(reportingApi, times(1)).fetchNewConfig();
verify(reportingApi, times(1)).fetchBlockedLists();
}

Expand Down
Loading