Skip to content

fix: Make FDv1 fallback TTL-based and always recover to FDv2 #542

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

Draft
beekld wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

fix: Make FDv1 fallback TTL-based and always recover to FDv2 #542

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
4 changes: 3 additions & 1 deletion ldclient/impl/datasourcev2/async_polling.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,8 @@
fdv1_polling_payload_to_changeset,
map_polling_result,
polling_payload_to_changeset,
polling_result_to_basis
polling_result_to_basis,
with_goodbye_fallback
)
from ldclient.impl.http import ASYNC_USER_AGENT
from ldclient.impl.util import (
Expand Down Expand Up @@ -228,6 +229,7 @@ async def fetch(self, selector: Optional[Selector]) -> PollingResult:
etag,
)

headers = with_goodbye_fallback(headers, data)
changeset_result = polling_payload_to_changeset(data)
if isinstance(changeset_result, _Success):
return _Success(value=(changeset_result.value, headers))
Expand Down
42 changes: 22 additions & 20 deletions ldclient/impl/datasourcev2/async_streaming.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,11 @@
with_fallback_signal
)
from ldclient.impl.datasystem import DiagnosticAccumulator, DiagnosticSource
from ldclient.impl.util import _LD_ENVID_HEADER, _LD_FD_FALLBACK_HEADER, log
from ldclient.impl.util import (
_LD_ENVID_HEADER,
fdv1_fallback_ttl_from_headers,
log
)
from ldclient.interfaces import (
AsyncSynchronizer,
ChangeSetBuilder,
Expand Down Expand Up @@ -153,12 +157,12 @@ async def sync(self, ss: SelectorStore) -> AsyncGenerator[Update, None]:
self._connection_attempt_start_time = time()

envid = None
# A Start action with the X-LD-FD-Fallback: true header sets fallback_requested.
# A Start action with the X-LD-FD-Fallback: true header sets fallback_ttl.
# We apply the current payload before halting, so consumers can serve that data
# while FDv1 takes over. Once set, the flag stays set: every later payload event
# while FDv1 takes over. Once set, it stays set: every later payload event
# or error re-emits the signal and halts the stream, even when that event has no
# directive header of its own.
fallback_requested = False
fallback_ttl: Optional[float] = None

try:
async for action in self._sse.all:
Expand All @@ -174,19 +178,18 @@ async def sync(self, ss: SelectorStore) -> AsyncGenerator[Update, None]:

(update, should_continue) = await self._handle_error(action.error, envid)
if update is not None:
yield with_fallback_signal(update, fallback_requested)
yield with_fallback_signal(update, fallback_ttl)

# The FDv1 Fallback Directive is one-way and terminal: if it
# was latched on a prior Start, we must not keep retrying the
# Once the FDv1 Fallback Directive was latched on a prior
# Start, we must not keep retrying the
# FDv2 endpoint even when the failure itself looks recoverable.
if fallback_requested or not should_continue:
if fallback_ttl is not None or not should_continue:
break
continue

if isinstance(action, Start) and action.headers is not None:
envid = action.headers.get(_LD_ENVID_HEADER, envid)
if action.headers.get(_LD_FD_FALLBACK_HEADER) == 'true':
fallback_requested = True
fallback_ttl = fdv1_fallback_ttl_from_headers(action.headers) or fallback_ttl

if not isinstance(action, Event):
continue
Expand All @@ -196,13 +199,12 @@ async def sync(self, ss: SelectorStore) -> AsyncGenerator[Update, None]:
if update is not None:
self._record_stream_init(False)
self._connection_attempt_start_time = None
if fallback_requested:
# The completed update is the natural moment to honor
# the latched directive: yield once with the signal,
# then halt — the consumer will switch to FDv1.
yield with_fallback_signal(update, fallback_requested)
break
update = with_fallback_signal(update, fallback_ttl)
yield update
# Honor a latched or in-band directive once the update is out:
# halt, and the consumer will switch to FDv1.
if update.fallback_to_fdv1:
break
except json.decoder.JSONDecodeError as e:
log.info(
"Error while handling stream event; will restart stream: %s", e
Expand All @@ -211,8 +213,8 @@ async def sync(self, ss: SelectorStore) -> AsyncGenerator[Update, None]:

(update, should_continue) = await self._handle_error(e, envid)
if update is not None:
yield with_fallback_signal(update, fallback_requested)
if fallback_requested or not should_continue:
yield with_fallback_signal(update, fallback_ttl)
if fallback_ttl is not None or not should_continue:
break
except Exception as e: # pylint: disable=broad-except
log.info(
Expand All @@ -227,8 +229,8 @@ async def sync(self, ss: SelectorStore) -> AsyncGenerator[Update, None]:
),
fallback_to_fdv1=False,
environment_id=envid,
), fallback_requested)
if fallback_requested:
), fallback_ttl)
if fallback_ttl is not None:
break
finally:
await self._sse.close()
Expand Down
4 changes: 3 additions & 1 deletion ldclient/impl/datasourcev2/polling.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,8 @@
fdv1_polling_payload_to_changeset,
map_polling_result,
polling_payload_to_changeset,
polling_result_to_basis
polling_result_to_basis,
with_goodbye_fallback
)
from ldclient.impl.http import HTTPFactory, _base_headers
from ldclient.impl.util import (
Expand Down Expand Up @@ -212,6 +213,7 @@ def fetch(self, selector: Optional[Selector]) -> PollingResult:
etag,
)

headers = with_goodbye_fallback(headers, data)
changeset_result = polling_payload_to_changeset(data)
if isinstance(changeset_result, _Success):
return _Success(value=(changeset_result.value, headers))
Expand Down
61 changes: 54 additions & 7 deletions ldclient/impl/datasourcev2/polling_common.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,20 +6,25 @@
from dataclasses import dataclass
from enum import Enum
from time import time
from typing import Mapping, Tuple
from typing import Mapping, Optional, Tuple

from urllib3._collections import HTTPHeaderDict

from ldclient.impl.datasystem.protocolv2 import (
DeleteObject,
EventName,
Goodbye,
PutObject
)
from ldclient.impl.util import (
_LD_ENVID_HEADER,
_LD_FD_FALLBACK_HEADER,
_LD_FD_FALLBACK_TTL_HEADER,
UnsuccessfulResponseException,
_Fail,
_Result,
_Success,
fdv1_fallback_ttl_from_headers,
http_error_message,
is_http_error_recoverable,
log
Expand All @@ -40,6 +45,31 @@
)


def with_goodbye_fallback(headers: Mapping, data: dict) -> Mapping:
"""
Return ``headers`` with the FDv1 fallback directive added when the polling
payload contains a goodbye event carrying protocolFallbackTTL, so the
in-band directive is handled like the X-LD-FD-Fallback header.
"""
events = data.get("events")
if not isinstance(events, list):
return headers

for event in events:
if not isinstance(event, dict) or event.get("event") != EventName.GOODBYE:
continue
ttl = event.get("data", {}).get("protocolFallbackTTL") if isinstance(event.get("data"), dict) else None
if ttl is None:
continue

merged = HTTPHeaderDict(headers)
merged[_LD_FD_FALLBACK_HEADER] = 'true'
merged[_LD_FD_FALLBACK_TTL_HEADER] = str(ttl) if isinstance(ttl, int) and not isinstance(ttl, bool) else ''
return merged

return headers


# pylint: disable=too-many-branches,too-many-return-statements
def polling_payload_to_changeset(data: dict) -> _Result[ChangeSet, str]:
"""
Expand All @@ -49,6 +79,7 @@ def polling_payload_to_changeset(data: dict) -> _Result[ChangeSet, str]:
return _Fail(error="Invalid payload: 'events' key is missing or not a list")

builder = ChangeSetBuilder()
goodbye_received = False

for event in data["events"]:
if not isinstance(event, dict):
Expand Down Expand Up @@ -83,6 +114,14 @@ def polling_payload_to_changeset(data: dict) -> _Result[ChangeSet, str]:
builder.add_delete(
delete_object.kind, delete_object.key, delete_object.version
)
elif event["event"] == EventName.GOODBYE:
try:
goodbye = Goodbye.from_dict(event["data"])
except ValueError as err:
return _Fail(error="Invalid JSON in goodbye", exception=err)

log.info("Polling response contained goodbye: %s", goodbye.reason)
goodbye_received = True
elif event["event"] == EventName.PAYLOAD_TRANSFERRED:
try:
selector = Selector.from_dict(event["data"])
Expand All @@ -94,6 +133,9 @@ def polling_payload_to_changeset(data: dict) -> _Result[ChangeSet, str]:
error="Invalid JSON in payload transferred object", exception=err
)

if goodbye_received:
return _Success(ChangeSetBuilder.no_changes())

return _Fail(error="didn't receive any known protocol events in polling payload")


Expand Down Expand Up @@ -170,11 +212,13 @@ def polling_result_to_basis(
if not isinstance(env_id, str):
env_id = None

fallback_ttl = fdv1_fallback_ttl_from_headers(headers)
basis = Basis(
change_set=change_set,
persist=change_set.selector.is_defined(),
environment_id=env_id,
fallback_to_fdv1=headers.get(_LD_FD_FALLBACK_HEADER) == 'true',
fallback_to_fdv1=fallback_ttl is not None,
fallback_ttl=fallback_ttl,
)

return _Success(value=basis)
Expand Down Expand Up @@ -212,11 +256,10 @@ def map_polling_result(
polling synchronizers.
"""
if isinstance(result, _Fail):
fallback = None
fallback_ttl = fdv1_fallback_ttl_from_headers(result.headers)
envid = None

if result.headers is not None:
fallback = result.headers.get(_LD_FD_FALLBACK_HEADER) == 'true'
envid = result.headers.get(_LD_ENVID_HEADER)

if isinstance(result.exception, UnsuccessfulResponseException):
Expand All @@ -229,13 +272,14 @@ def map_polling_result(
),
)

if fallback:
if fallback_ttl is not None:
return PollDecision(
Update(
state=DataSourceState.OFF,
error=error_info,
fallback_to_fdv1=True,
environment_id=envid,
fallback_ttl=fallback_ttl,
),
PollAction.BREAK,
)
Expand Down Expand Up @@ -269,13 +313,14 @@ def map_polling_result(

# Even a non-HTTP error (e.g. malformed JSON) can carry the fallback
# header. If so, halt rather than retrying the FDv2 endpoint.
if fallback:
if fallback_ttl is not None:
return PollDecision(
Update(
state=DataSourceState.OFF,
error=error_info,
fallback_to_fdv1=True,
environment_id=envid,
fallback_ttl=fallback_ttl,
),
PollAction.BREAK,
)
Expand All @@ -290,12 +335,14 @@ def map_polling_result(
)

(change_set, headers) = result.value
fallback_ttl = fdv1_fallback_ttl_from_headers(headers)
return PollDecision(
Update(
state=DataSourceState.VALID,
change_set=change_set,
environment_id=headers.get(_LD_ENVID_HEADER),
fallback_to_fdv1=headers.get(_LD_FD_FALLBACK_HEADER) == 'true',
fallback_to_fdv1=fallback_ttl is not None,
fallback_ttl=fallback_ttl,
),
PollAction.WAIT_BREAK_IF_SET,
)
47 changes: 24 additions & 23 deletions ldclient/impl/datasourcev2/streaming.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,11 @@
)
from ldclient.impl.datasystem import DiagnosticAccumulator, DiagnosticSource
from ldclient.impl.http import HTTPFactory, _base_headers
from ldclient.impl.util import _LD_ENVID_HEADER, _LD_FD_FALLBACK_HEADER, log
from ldclient.impl.util import (
_LD_ENVID_HEADER,
fdv1_fallback_ttl_from_headers,
log
)
from ldclient.interfaces import (
ChangeSetBuilder,
DataSourceErrorInfo,
Expand Down Expand Up @@ -161,14 +165,13 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]:
self._connection_attempt_start_time = time()

envid = None
# fallback_requested is set when a Start action carries
# fallback_ttl is set when a Start action carries
# X-LD-FD-Fallback: true. We finish applying the current payload
# before halting, so consumers can serve the server-provided data
# while FDv1 takes over. The latch is one-way and terminal: once
# set, any subsequent payload-completing event or error must carry
# the signal forward and halt the stream, even if the failure path
# itself doesn't see the directive header.
fallback_requested = False
# while FDv1 takes over. Once set, any subsequent payload-completing
# event or error must carry the signal forward and halt the stream,
# even if the failure path itself doesn't see the directive header.
fallback_ttl: Optional[float] = None

for action in self._sse.all:
if isinstance(action, Fault):
Expand All @@ -183,19 +186,18 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]:

(update, should_continue) = self._handle_error(action.error, envid)
if update is not None:
yield with_fallback_signal(update, fallback_requested)
yield with_fallback_signal(update, fallback_ttl)

# The FDv1 Fallback Directive is one-way and terminal: if it
# was latched on a prior Start, we must not keep retrying the
# Once the FDv1 Fallback Directive was latched on a prior
# Start, we must not keep retrying the
# FDv2 endpoint even when the failure itself looks recoverable.
if fallback_requested or not should_continue:
if fallback_ttl is not None or not should_continue:
break
continue

if isinstance(action, Start) and action.headers is not None:
envid = action.headers.get(_LD_ENVID_HEADER, envid)
if action.headers.get(_LD_FD_FALLBACK_HEADER) == 'true':
fallback_requested = True
fallback_ttl = fdv1_fallback_ttl_from_headers(action.headers) or fallback_ttl

if not isinstance(action, Event):
continue
Expand All @@ -205,13 +207,12 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]:
if update is not None:
self._record_stream_init(False)
self._connection_attempt_start_time = None
if fallback_requested:
# The completed update is the natural moment to honor
# the latched directive: yield once with the signal,
# then halt — the consumer will switch to FDv1.
yield with_fallback_signal(update, fallback_requested)
break
update = with_fallback_signal(update, fallback_ttl)
yield update
# Honor a latched or in-band directive once the update is out:
# halt, and the consumer will switch to FDv1.
if update.fallback_to_fdv1:
break
except json.decoder.JSONDecodeError as e:
log.info(
"Error while handling stream event; will restart stream: %s", e
Expand All @@ -220,8 +221,8 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]:

(update, should_continue) = self._handle_error(e, envid)
if update is not None:
yield with_fallback_signal(update, fallback_requested)
if fallback_requested or not should_continue:
yield with_fallback_signal(update, fallback_ttl)
if fallback_ttl is not None or not should_continue:
break
except Exception as e: # pylint: disable=broad-except
log.info(
Expand All @@ -236,8 +237,8 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]:
),
fallback_to_fdv1=False,
environment_id=envid,
), fallback_requested)
if fallback_requested:
), fallback_ttl)
if fallback_ttl is not None:
break

self._sse.close()
Expand Down
Loading
Loading