diff --git a/fm-provider-test-harness b/fm-provider-test-harness index ee15379..e3fbb5d 160000 --- a/fm-provider-test-harness +++ b/fm-provider-test-harness @@ -1 +1 @@ -Subproject commit ee153794814ac3526e3e80dc30eb4ad24c82f14b +Subproject commit e3fbb5d488d7f7ad51b885974a08f93b302cc4f6 diff --git a/src/main/java/com/dynatrace/openfeature/DynatraceProvider.java b/src/main/java/com/dynatrace/openfeature/DynatraceProvider.java index 8a310d3..37bdead 100644 --- a/src/main/java/com/dynatrace/openfeature/DynatraceProvider.java +++ b/src/main/java/com/dynatrace/openfeature/DynatraceProvider.java @@ -66,14 +66,6 @@ public final class DynatraceProvider extends EventProvider { static final Duration GRACE_PERIOD = Duration.ofMinutes(1); - /** - * How long to wait after an SSE fault before treating it as a real disconnect. The underlying - * event source auto-reconnects and fires its error/closed callbacks on every transient blip, - * so escalating immediately would flap STALE/READY on routine reconnects. We debounce: if the - * stream reopens within this window it was a blip (no-op); otherwise we escalate. - */ - static final Duration SSE_DISCONNECT_DEBOUNCE = Duration.ofSeconds(5); - /** SDK key format: {token}.server_{geo}_{publicRandom}.{secret}_{sha1} */ static final Pattern KEY_PATTERN = Pattern.compile( "^(dt[0-9]{2})\\.(server)_([a-z]{2,})_([a-zA-Z0-9]{10,})\\.([a-fA-F0-9]{32,})_([a-f0-9]{10,})$"); @@ -87,7 +79,6 @@ public final class DynatraceProvider extends EventProvider { private final Duration gracePeriod; private final Duration pollIntervalConnected; private final Duration pollIntervalDisconnected; - private final Duration sseDisconnectDebounce; private final DynatraceMetricsAggregator metricsAggregator; private final EvaluationMetricsFlusher metricsFlusher; @@ -99,7 +90,7 @@ public final class DynatraceProvider extends EventProvider { private final AtomicReference state = new AtomicReference<>(ProviderState.NOT_READY); - /** True while the SSE connection is considered healthy (stays true through brief blips). */ + /** True while the SSE stream is open; only selects the poll cadence. */ private final AtomicBoolean sseConnected = new AtomicBoolean(false); private volatile SseManager sseManager; @@ -112,9 +103,6 @@ public final class DynatraceProvider extends EventProvider { private volatile ScheduledExecutorService graceScheduler; private volatile ScheduledFuture graceTimer; - // debounce timer: escalates a sustained SSE fault to a real disconnect (see SSE_DISCONNECT_DEBOUNCE) - private volatile ScheduledFuture sseDisconnectTimer; - /** * Creates a provider from the SDK key. CDN URL is derived automatically from the key. * @@ -139,27 +127,21 @@ public DynatraceProvider(String sdkKey, DynatraceProviderOptions options) { // for testing DynatraceProvider(String sdkKey, String cdnUrl, FlagdCore flagdCore, CdnFetcher fetcher) { this(sdkKey, null, cdnUrl, flagdCore, fetcher, buildDefaultHttpClient(), - POLL_INTERVAL_SSE_CONNECTED, POLL_INTERVAL_SSE_DISCONNECTED, GRACE_PERIOD, SSE_DISCONNECT_DEBOUNCE); + POLL_INTERVAL_SSE_CONNECTED, POLL_INTERVAL_SSE_DISCONNECTED, GRACE_PERIOD); } // for testing: a single fixed poll interval for both SSE states (cadence independent of SSE) DynatraceProvider(String sdkKey, String cdnUrl, FlagdCore flagdCore, CdnFetcher fetcher, Duration pollInterval, Duration gracePeriod) { - this(sdkKey, cdnUrl, flagdCore, fetcher, pollInterval, gracePeriod, SSE_DISCONNECT_DEBOUNCE); - } - - // for testing: fixed poll interval plus an explicit SSE-disconnect debounce - DynatraceProvider(String sdkKey, String cdnUrl, FlagdCore flagdCore, CdnFetcher fetcher, - Duration pollInterval, Duration gracePeriod, Duration sseDisconnectDebounce) { this(sdkKey, null, cdnUrl, flagdCore, fetcher, buildDefaultHttpClient(), - pollInterval, pollInterval, gracePeriod, sseDisconnectDebounce); + pollInterval, pollInterval, gracePeriod); } // tests only: lets tests inject every collaborator individually DynatraceProvider(String sdkKey, DynatraceProviderOptions options, FlagdCore flagdCore, HttpClient httpClient) { this(sdkKey, options, buildCdnUrl(sdkKey, options), flagdCore, new CdnFetcher(buildCdnUrl(sdkKey, options), httpClient), httpClient, - POLL_INTERVAL_SSE_CONNECTED, POLL_INTERVAL_SSE_DISCONNECTED, GRACE_PERIOD, SSE_DISCONNECT_DEBOUNCE); + POLL_INTERVAL_SSE_CONNECTED, POLL_INTERVAL_SSE_DISCONNECTED, GRACE_PERIOD); } // tests only: production wiring with an overridden SSE-connected poll interval and grace period @@ -172,28 +154,28 @@ public DynatraceProvider(String sdkKey, DynatraceProviderOptions options) { private DynatraceProvider(String sdkKey, DynatraceProviderOptions options, String cdnUrl, FlagdCore flagdCore, HttpClient httpClient, Duration pollIntervalConnected, Duration gracePeriod) { this(sdkKey, options, cdnUrl, flagdCore, new CdnFetcher(cdnUrl, httpClient), httpClient, - pollIntervalConnected, POLL_INTERVAL_SSE_DISCONNECTED, gracePeriod, SSE_DISCONNECT_DEBOUNCE); + pollIntervalConnected, POLL_INTERVAL_SSE_DISCONNECTED, gracePeriod); } // tests only: as above, plus a pre-built CdnFetcher and timing controls (fixed cadence both states) DynatraceProvider(String sdkKey, DynatraceProviderOptions options, FlagdCore flagdCore, CdnFetcher fetcher, Duration pollInterval, Duration gracePeriod) { this(sdkKey, options, buildCdnUrl(sdkKey, options), flagdCore, fetcher, buildDefaultHttpClient(), - pollInterval, pollInterval, gracePeriod, SSE_DISCONNECT_DEBOUNCE); + pollInterval, pollInterval, gracePeriod); } // for testing: distinct connected/disconnected cadences, to observe the SSE-transition consequence DynatraceProvider(String sdkKey, String cdnUrl, FlagdCore flagdCore, CdnFetcher fetcher, Duration pollIntervalConnected, Duration pollIntervalDisconnected, - Duration gracePeriod, Duration sseDisconnectDebounce) { + Duration gracePeriod) { this(sdkKey, null, cdnUrl, flagdCore, fetcher, buildDefaultHttpClient(), - pollIntervalConnected, pollIntervalDisconnected, gracePeriod, sseDisconnectDebounce); + pollIntervalConnected, pollIntervalDisconnected, gracePeriod); } private DynatraceProvider(String sdkKey, DynatraceProviderOptions options, String cdnUrl, FlagdCore flagdCore, CdnFetcher fetcher, HttpClient metricsHttpClient, Duration pollIntervalConnected, Duration pollIntervalDisconnected, - Duration gracePeriod, Duration sseDisconnectDebounce) { + Duration gracePeriod) { String trimmedSdkKey = validateSdkKey(sdkKey).group(0); DynatraceProviderOptions safeOptions = options == null ? DynatraceProviderOptions.builder().build() : options; this.cdnUrl = cdnUrl; @@ -203,7 +185,6 @@ private DynatraceProvider(String sdkKey, DynatraceProviderOptions options, Strin this.pollIntervalConnected = pollIntervalConnected; this.pollIntervalDisconnected = pollIntervalDisconnected; this.gracePeriod = gracePeriod; - this.sseDisconnectDebounce = sseDisconnectDebounce; this.metricsAggregator = new DynatraceMetricsAggregator(safeOptions.getMetricsHardCapDistinctFlags()); this.metricsFlusher = new EvaluationMetricsFlusher( metricsAggregator, @@ -267,16 +248,17 @@ public List getProviderHooks() { } /** - * Fetches the initial configuration from the CDN, opens the SSE connection, starts polling, + * Fetches the initial configuration from the CDN, starts polling, opens the SSE connection, * and starts the evaluation-metrics flusher. * *

Initialization order (per spec 3.1): *

    *
  1. Fetch flagd config JSON from CDN
    2. *
  2. Load config into flagd-core
    3. - *
  3. Open SSE connection (async; errors handled by callbacks)
    4. *
  4. Start polling timer, only once step 1 has completed, so the initial fetch never * contends with the first tick (spec 2.1.1)
    5. + *
  5. Open SSE connection (async; errors handled by callbacks), only once polling has + * started, so the catch-up fetch on connect has a manager to run on
    6. *
  6. Emit {@code PROVIDER_READY} (via return)
    7. *
* @@ -302,6 +284,7 @@ public void initialize(EvaluationContext evaluationContext) throws Exception { }); RuntimeException initFailure = null; + String initialConfig = null; try { FetchResult result; try { @@ -330,9 +313,7 @@ public void initialize(EvaluationContext evaluationContext) throws Exception { } state.set(ProviderState.READY); - - // open SSE connection (spec 3.1 step 3) - ensureSseStarted(body); + initialConfig = body; } catch (RuntimeException e) { // Don't rethrow yet: polling below recovers, and invalidating stops a phantom 304 // from "recovering" to READY with no flags loaded. @@ -353,13 +334,12 @@ public void initialize(EvaluationContext evaluationContext) throws Exception { this::onFetchFailure, pollIntervalDisconnected); pollingManager.start(); - // If SSE connected during init (before the manager existed), adopt the connected cadence. - if (sseConnected.get()) { - pollingManager.setInterval(pollIntervalConnected); - } - metricsFlusher.start(); + if (initialConfig != null) { + ensureSseStarted(initialConfig); + } + if (initFailure != null) { throw initFailure; } @@ -445,7 +425,6 @@ public void shutdown() { pm.stop(); } cancelGraceTimer(); - cancelSseDisconnectTimer(); ScheduledExecutorService gs = graceScheduler; if (gs != null) { gs.shutdown(); @@ -460,72 +439,27 @@ public void shutdown() { log.info("Dynatrace provider shut down"); } - // -- SSE callbacks (package-private for testing) -- - + // SSE only selects the poll cadence; provider state follows CDN fetch outcomes alone synchronized void onSseConnected() { - // A (re)connect aborts any pending disconnect escalation. If one was pending, the stream - // just blipped, so pull current config in case we missed a change while it was down. - boolean hadPendingDisconnect = cancelSseDisconnectTimer(); - boolean wasConnected = sseConnected.getAndSet(true); + sseConnected.set(true); PollingManager pm = pollingManager; - if (pm != null) { - pm.setInterval(pollIntervalConnected); - } - if (!wasConnected) { - log.info("SSE connected; poll interval switched to {}", pollIntervalConnected); - } - if (pm != null && (!wasConnected || hadPendingDisconnect)) { - // Reconnect catch-up: no announced lastModified, so this re-fetch is never dropped. - pm.requestOutOfBandFetch(null); - } - } - - /** - * SSE fault callback. The event source auto-reconnects and fires this on every transient blip, - * so we don't escalate immediately: we start a debounce timer (idempotent across repeat faults) - * and only treat it as a real disconnect if the stream hasn't reopened by the time it fires. - */ - synchronized void onSseDisconnected() { - if (!sseConnected.get()) { - // already escalated to disconnected; nothing to do - return; - } - if (sseDisconnectTimer != null && !sseDisconnectTimer.isDone()) { - // debounce already running; this is a repeat fault callback + if (pm == null) { return; } - log.debug("SSE fault; starting {}ms disconnect debounce", sseDisconnectDebounce.toMillis()); - try { - sseDisconnectTimer = graceScheduler.schedule( - this::escalateSseDisconnect, - sseDisconnectDebounce.toMillis(), - TimeUnit.MILLISECONDS); - } catch (RejectedExecutionException e) { - // scheduler already shut down; ignore - } + log.info("SSE connected"); + // catch up on anything announced while the stream was down; never dropped + pm.requestOutOfBandFetch(null); } - /** - * Runs when an SSE fault has persisted past the debounce window: switch to the aggressive poll - * cadence and go STALE. Switching the interval re-anchors the ticker, so the first fast poll is - * one (now short) interval away rather than up to a full SSE-connected interval, which closes - * the cadence gap without an extra out-of-band fetch that would instantly flap STALE to READY. - */ - private synchronized void escalateSseDisconnect() { + synchronized void onSseDisconnected() { if (!sseConnected.getAndSet(false)) { - return; // a reconnect beat us to it + return; } - log.warn("SSE disconnected; increasing poll frequency to {}", pollIntervalDisconnected); + log.warn("SSE disconnected; poll interval switched to {}", pollIntervalDisconnected); PollingManager pm = pollingManager; if (pm != null) { pm.setInterval(pollIntervalDisconnected); } - if (state.compareAndSet(ProviderState.READY, ProviderState.STALE)) { - emitProviderStale(ProviderEventDetails.builder() - .message("SSE disconnected") - .build()); - startGraceTimer(); - } } /** The poll/fetch coordinator, or null before {@link #initialize} has run. */ @@ -533,17 +467,6 @@ PollingManager pollingManager() { return pollingManager; } - private synchronized boolean cancelSseDisconnectTimer() { - ScheduledFuture timer = sseDisconnectTimer; - if (timer == null) { - return false; - } - boolean wasPending = !timer.isDone(); - timer.cancel(false); - sseDisconnectTimer = null; - return wasPending; - } - /** * Handles a parsed SSE message per the SSE message format spec: *