From a42feedac5dda34de9e57852325b4264d8257128 Mon Sep 17 00:00:00 2001 From: Deep Santoshwar Date: Sat, 8 Aug 2026 05:39:04 +0530 Subject: [PATCH 1/2] fix(environments): guard environment document writes against unseeded feature states MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The environment document write is driven by the audit log post_save signal, which is enqueued before `create_feature_states` has seeded the initial feature states for a new environment or feature. When the write won the race, edge-api served a document missing those states. For environments using v2 feature versioning the document never recovered on its own: seeded feature states are excluded from the audit log, so no later audit record triggered a corrective write, and the document stayed partial until it was rebuilt by hand. Mark that window with `Environment.is_creating` (extended from clones to standard creation) and a new `Feature.is_creating`, and have `process_environment_update` re-enqueue itself with exponential backoff while either is set, giving up after three attempts so a stuck flag cannot block an environment's updates forever. This replaces the one second `delay_until` shim, which was a timer rather than a guarantee. Both flags are cleared with a direct row update, so their historical records are always stale — keep `is_creating` out of the feature's history, and out of the environment's audit log change details, where a spurious entry was already shown for cloned environments. Partial writes were also invisible: add a `flagsmith_environment_document_writes_total` counter and an `environment_document.written` event carrying the feature state count, so a truncated document can be alerted on rather than reported through support. --- api/core/signals.py | 23 +--- api/environments/constants.py | 6 + .../dynamodb/wrappers/environment_wrapper.py | 84 ++++++++---- api/environments/metrics.py | 7 + api/environments/models.py | 17 ++- api/environments/tasks.py | 57 +++++++- .../0068_add_feature_is_creating.py | 21 +++ api/features/models.py | 23 +++- .../test_unit_dynamo_environment_wrapper.py | 51 +++++++ ...st_unit_dynamodb_environment_v2_wrapper.py | 8 ++ .../test_unit_environments_models.py | 27 ++++ .../test_unit_environments_tasks.py | 127 ++++++++++++++++++ .../features/test_unit_features_models.py | 28 ++++ api/util/dataclasses.py | 1 + api/util/mappers/dynamodb.py | 2 + .../observability/_events-catalogue.md | 24 +++- .../observability/_metrics-catalogue.md | 9 ++ 17 files changed, 462 insertions(+), 53 deletions(-) create mode 100644 api/features/migrations/0068_add_feature_is_creating.py diff --git a/api/core/signals.py b/api/core/signals.py index c80d8fbef0ad..5dc158ef38d6 100644 --- a/api/core/signals.py +++ b/api/core/signals.py @@ -1,10 +1,7 @@ import logging -from django.conf import settings from django.core.exceptions import ObjectDoesNotExist -from django.utils import timezone from simple_history.models import HistoricalRecords # type: ignore[import-untyped] -from task_processor.task_run_method import TaskRunMethod from audit import tasks from core.models import AbstractBaseAuditableModel @@ -19,21 +16,10 @@ def create_audit_log_from_historical_record( # type: ignore[no-untyped-def] history_instance, **kwargs, ): - # The environment document in dynamodb is updated based on the post_save signal from the audit log - # When creating a new feature, the feature states are created after the feature has been created. - # i.e: the below task gets created/scheduled before feature states are created - # Usually, there is enough time for the main thread to create the feature states - # before the task is executed, but not always. - # In those cases, we send the environment document to dynamodb without any feature states for the new feature. - # In order to avoid this, either we need to update environment - # document when creating feature states - # or delay the execution of this task - # We prefer to delay the execution of the task because of it's low surface area - delay_until = ( - timezone.now() + timezone.timedelta(seconds=1) # type: ignore[attr-defined] - if settings.TASK_RUN_METHOD == TaskRunMethod.TASK_PROCESSOR - else None - ) + # Note: this task can run before the feature states of a newly created feature or + # environment exist. `environments.tasks.process_environment_update` guards the + # environment document write against that, using `Feature.is_creating` and + # `Environment.is_creating`. if instance.get_skip_create_audit_log(): return @@ -62,7 +48,6 @@ def create_audit_log_from_historical_record( # type: ignore[no-untyped-def] "history_user_id": getattr(history_user, "id", None), "history_record_class_path": instance.history_record_class_path, }, - delay_until=delay_until, ) diff --git a/api/environments/constants.py b/api/environments/constants.py index 8f83d38c5b3e..1064a4baa8c6 100644 --- a/api/environments/constants.py +++ b/api/environments/constants.py @@ -1,3 +1,9 @@ +# How many times an environment document write is deferred while the environments or +# features it covers are still being created, and the base of the exponential backoff +# (in seconds) between those attempts, i.e. 1s, 2s, 4s. +ENVIRONMENT_DOCUMENT_WRITE_MAX_DEFERRALS = 3 +ENVIRONMENT_DOCUMENT_WRITE_DEFERRAL_SECONDS = 1 + IDENTITY_INTEGRATIONS_RELATION_NAMES = [ "amplitude_config", "heap_config", diff --git a/api/environments/dynamodb/wrappers/environment_wrapper.py b/api/environments/dynamodb/wrappers/environment_wrapper.py index 8abe96a89d2b..e1bbf86a2e52 100644 --- a/api/environments/dynamodb/wrappers/environment_wrapper.py +++ b/api/environments/dynamodb/wrappers/environment_wrapper.py @@ -21,6 +21,7 @@ from environments.metrics import ( flagsmith_dynamo_environment_document_compression_ratio, flagsmith_dynamo_environment_document_size_bytes, + flagsmith_environment_document_writes_total, ) from integrations.flagsmith.client import get_openfeature_client from util.mappers import ( @@ -71,37 +72,62 @@ def _write_environments(self, environments: Iterable["Environment"]) -> None: ) assert self.table - with self.table.batch_writer() as writer: - for environment in environments: - organisation = environment.project.organisation - if openfeature_client.get_boolean_value( - "compress_dynamo_documents", - default_value=False, - evaluation_context=organisation.openfeature_evaluation_context, - ): - result = self._map_compressed_environment_document(environment) - writer.put_item(Item=result.document) - - flagsmith_dynamo_environment_document_size_bytes.labels( - table=self.get_table_name(), - compressed="true", - ).observe(result.compressed_size_bytes) - flagsmith_dynamo_environment_document_compression_ratio.labels( - table=self.get_table_name(), - ).observe(result.compression_ratio) + written = 0 + try: + with self.table.batch_writer() as writer: + for environment in environments: + organisation = environment.project.organisation + if openfeature_client.get_boolean_value( + "compress_dynamo_documents", + default_value=False, + evaluation_context=organisation.openfeature_evaluation_context, + ): + result = self._map_compressed_environment_document(environment) + writer.put_item(Item=result.document) + + flagsmith_dynamo_environment_document_size_bytes.labels( + table=self.get_table_name(), + compressed="true", + ).observe(result.compressed_size_bytes) + flagsmith_dynamo_environment_document_compression_ratio.labels( + table=self.get_table_name(), + ).observe(result.compression_ratio) + logger.info( + "environment-document-compressed", + environment_id=environment.id, + environment_api_key=environment.api_key, + ) + document_bytes = result.compressed_size_bytes + feature_states_count = result.feature_states_count + else: + item = self._map_environment_document(environment) + writer.put_item(Item=item) + + document_bytes = estimate_document_size(item) + feature_states_count = len(item["feature_states"]) + flagsmith_dynamo_environment_document_size_bytes.labels( + table=self.get_table_name(), + compressed="false", + ).observe(document_bytes) + + written += 1 logger.info( - "environment-document-compressed", - environment_id=environment.id, - environment_api_key=environment.api_key, + "environment_document.written", + environment__id=environment.id, + feature_states__count=feature_states_count, + document__bytes=document_bytes, ) - else: - item = self._map_environment_document(environment) - writer.put_item(Item=item) - - flagsmith_dynamo_environment_document_size_bytes.labels( - table=self.get_table_name(), - compressed="false", - ).observe(estimate_document_size(item)) + except Exception: + # The batch writer flushes on exit, so a failure can cover documents + # already handed to it as well as the one being written. + flagsmith_environment_document_writes_total.labels(result="failure").inc( + max(written, 1) + ) + raise + + flagsmith_environment_document_writes_total.labels(result="success").inc( + written + ) class DynamoEnvironmentWrapper(BaseDynamoEnvironmentWrapper): diff --git a/api/environments/metrics.py b/api/environments/metrics.py index 3e41e75fd83e..44a592e03f42 100644 --- a/api/environments/metrics.py +++ b/api/environments/metrics.py @@ -14,6 +14,13 @@ ["result"], ) +flagsmith_environment_document_writes_total = prometheus_client.Counter( + "flagsmith_environment_document_writes_total", + "Environment documents written to DynamoDB. `result` label is either `success` or " + "`failure`.", + ["result"], +) + flagsmith_dynamo_environment_document_size_bytes = prometheus_client.Histogram( "flagsmith_dynamo_environment_document_size_bytes", "Size of environment documents written to DynamoDB.", diff --git a/api/environments/models.py b/api/environments/models.py index 80f79ef08889..2bf75bf59006 100644 --- a/api/environments/models.py +++ b/api/environments/models.py @@ -17,6 +17,7 @@ AFTER_DELETE, AFTER_SAVE, AFTER_UPDATE, + BEFORE_CREATE, BEFORE_SAVE, LifecycleModel, hook, @@ -82,7 +83,9 @@ class Environment( LifecycleModel, # type: ignore[misc] abstract_base_auditable_model_factory( # type: ignore[misc] - change_details_excluded_fields=["updated_at"], + # `is_creating` is an internal lifecycle flag, and is cleared without a + # historical record, so its recorded value is always stale. + change_details_excluded_fields=["updated_at", "is_creating"], historical_records_excluded_fields=["uuid"], ), SoftDeleteObject, # type: ignore[misc] @@ -180,9 +183,21 @@ class Environment( class Meta: ordering = ["id"] + @hook(BEFORE_CREATE) # type: ignore[misc] + def mark_as_creating(self) -> None: + # The environment row is committed before `create_feature_states` seeds its + # initial feature states, so anything reading the environment in between sees + # it without any flags. `is_creating` marks that window for the environment + # document writer. Clones set the flag in `clone()` instead, since create + # hooks don't fire for them. + self.is_creating = True + @hook(AFTER_CREATE) # type: ignore[misc] def create_feature_states(self) -> None: FeatureState.create_initial_feature_states_for_environment(environment=self) + self.is_creating = False + # Update the row directly to avoid re-triggering hooks for a lifecycle flag. + Environment.objects.filter(id=self.id).update(is_creating=False) @hook(AFTER_UPDATE) # type: ignore[misc] def clear_environment_cache(self) -> None: diff --git a/api/environments/tasks.py b/api/environments/tasks.py index 09a8becb3b30..5906a953b01f 100644 --- a/api/environments/tasks.py +++ b/api/environments/tasks.py @@ -1,17 +1,27 @@ +from datetime import timedelta + +import structlog +from django.conf import settings from django.db.models import Prefetch, Q from django.utils import timezone from task_processor.decorators import ( register_task_handler, ) from task_processor.models import TaskPriority +from task_processor.task_run_method import TaskRunMethod from audit.models import AuditLog +from environments.constants import ( + ENVIRONMENT_DOCUMENT_WRITE_DEFERRAL_SECONDS, + ENVIRONMENT_DOCUMENT_WRITE_MAX_DEFERRALS, +) from environments.dynamodb import DynamoIdentityWrapper from environments.models import ( Environment, environment_v2_wrapper, environment_wrapper, ) +from features.models import Feature from features.multivariate.models import MultivariateFeatureStateValue from features.versioning.models import EnvironmentFeatureVersion from features.versioning.versioning_service import ( @@ -22,16 +32,61 @@ send_environment_update_message_for_project, ) +logger = structlog.get_logger("environments") + @register_task_handler(priority=TaskPriority.HIGH) def rebuild_environment_document(environment_id: int) -> None: Environment.write_environment_documents(environment_id=environment_id) +def _is_seeding_feature_states(audit_log: AuditLog) -> bool: + """ + Whether any environment or feature the audit log covers is still seeding its + initial feature states, and would therefore be written to DynamoDB incomplete. + """ + environments_filter = Q(project_id=audit_log.project_id) + if audit_log.environment_id: + environments_filter &= Q(id=audit_log.environment_id) + + return bool( + Environment.objects.filter(environments_filter, is_creating=True).exists() + or Feature.objects.filter( + project_id=audit_log.project_id, is_creating=True + ).exists() + ) + + @register_task_handler(priority=TaskPriority.HIGHEST) -def process_environment_update(audit_log_id: int): # type: ignore[no-untyped-def] +def process_environment_update(audit_log_id: int, deferrals: int = 0): # type: ignore[no-untyped-def] audit_log = AuditLog.objects.get(id=audit_log_id) + # Deferring relies on `delay_until`, which is a no-op without the task processor — + # re-enqueueing there would drop the write and leave the document stale. + if ( + settings.TASK_RUN_METHOD == TaskRunMethod.TASK_PROCESSOR + and _is_seeding_feature_states(audit_log) + ): + if deferrals < ENVIRONMENT_DOCUMENT_WRITE_MAX_DEFERRALS: + process_environment_update.delay( + kwargs={"audit_log_id": audit_log_id, "deferrals": deferrals + 1}, + delay_until=timezone.now() + + timedelta( + seconds=ENVIRONMENT_DOCUMENT_WRITE_DEFERRAL_SECONDS * 2**deferrals + ), + ) + return + + # Writing a document that may be missing feature states is preferable to not + # writing one at all, since a stuck `is_creating` flag would otherwise stop + # every future update to this environment. + logger.warning( + "environment_document.written_while_seeding", + audit_log__id=audit_log_id, + environment__id=audit_log.environment_id, + project__id=audit_log.project_id, + ) + # Send environment document to dynamodb Environment.write_environment_documents( environment_id=audit_log.environment_id, project_id=audit_log.project_id diff --git a/api/features/migrations/0068_add_feature_is_creating.py b/api/features/migrations/0068_add_feature_is_creating.py new file mode 100644 index 000000000000..f9bf51466512 --- /dev/null +++ b/api/features/migrations/0068_add_feature_is_creating.py @@ -0,0 +1,21 @@ +# Generated by Django 5.2.16 on 2026-08-07 23:55 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ("features", "0067_add_feature_state_mv_hashing_salt"), + ] + + operations = [ + migrations.AddField( + model_name="feature", + name="is_creating", + field=models.BooleanField( + default=False, + help_text="Attribute used to indicate when a feature is still being created, and its initial feature states are not yet guaranteed to exist", + ), + ), + ] diff --git a/api/features/models.py b/api/features/models.py index eaaf75074503..630ae4b32f61 100644 --- a/api/features/models.py +++ b/api/features/models.py @@ -92,7 +92,9 @@ class Feature( # type: ignore[django-manager-missing] SoftDeleteExportableModel, CustomLifecycleModelMixin, - abstract_base_auditable_model_factory(["uuid"]), # type: ignore[misc] + # `is_creating` is an internal lifecycle flag, and is cleared without a historical + # record, so it is kept out of the feature's history entirely. + abstract_base_auditable_model_factory(["uuid", "is_creating"]), # type: ignore[misc] ): name = models.CharField(max_length=2000) created_date = models.DateTimeField("DateCreated", auto_now_add=True) @@ -128,6 +130,14 @@ class Feature( # type: ignore[django-manager-missing] is_server_key_only = models.BooleanField(default=False) + is_creating = models.BooleanField( + default=False, + help_text=( + "Attribute used to indicate when a feature is still being created, and its" + " initial feature states are not yet guaranteed to exist" + ), + ) + history_record_class_path = "features.models.HistoricalFeature" related_object_type = RelatedObjectType.FEATURE @@ -178,9 +188,20 @@ def create_gitlab_comment(self) -> None: args=(self.name, self.id, self.project_id), ) + @hook(BEFORE_CREATE) # type: ignore[misc] + def mark_as_creating(self) -> None: + # The feature row is committed before `create_feature_states` seeds its feature + # states, so anything reading the project in between sees a feature with no + # states in any environment. `is_creating` marks that window for the + # environment document writer. + self.is_creating = True + @hook(AFTER_CREATE) def create_feature_states(self): # type: ignore[no-untyped-def] FeatureState.create_initial_feature_states_for_feature(feature=self) + self.is_creating = False + # Update the row directly to avoid re-triggering hooks for a lifecycle flag. + Feature.objects.filter(id=self.id).update(is_creating=False) @hook(AFTER_SAVE) # type: ignore[misc] def delete_identity_overrides(self) -> None: diff --git a/api/tests/unit/environments/dynamodb/wrappers/test_unit_dynamo_environment_wrapper.py b/api/tests/unit/environments/dynamodb/wrappers/test_unit_dynamo_environment_wrapper.py index 6f9010ecb411..84375c3ae009 100644 --- a/api/tests/unit/environments/dynamodb/wrappers/test_unit_dynamo_environment_wrapper.py +++ b/api/tests/unit/environments/dynamodb/wrappers/test_unit_dynamo_environment_wrapper.py @@ -1,8 +1,11 @@ +from unittest.mock import ANY + import pytest from boto3.dynamodb.types import Binary from common.test_tools import AssertMetricFixture from django.core.exceptions import ObjectDoesNotExist from mypy_boto3_dynamodb.service_resource import Table +from pytest_mock import MockerFixture from pytest_structlog import StructuredLogCapture from environments.dynamodb import DynamoEnvironmentWrapper @@ -99,6 +102,47 @@ def test_write_environments__uncompressed__observes_size_metric( ) +def test_write_environments__successful_write__counts_success( + environment: Environment, + dynamo_environment_wrapper: DynamoEnvironmentWrapper, + flagsmith_environment_table: Table, + assert_metric: AssertMetricFixture, +) -> None: + # Given / When + dynamo_environment_wrapper.write_environments([environment]) + + # Then + assert_metric( + name="flagsmith_environment_document_writes_total", + labels={"result": "success"}, + value=1.0, + ) + + +def test_write_environments__failed_write__counts_failure( + environment: Environment, + assert_metric: AssertMetricFixture, + mocker: MockerFixture, +) -> None: + # Given + dynamo_environment_wrapper = DynamoEnvironmentWrapper() + mocked_dynamo_table = mocker.patch.object(dynamo_environment_wrapper, "_table") + mocked_dynamo_table.batch_writer.return_value.__enter__.return_value.put_item.side_effect = RuntimeError( + "DynamoDB unavailable" + ) + + # When + with pytest.raises(RuntimeError): + dynamo_environment_wrapper.write_environments([environment]) + + # Then + assert_metric( + name="flagsmith_environment_document_writes_total", + labels={"result": "failure"}, + value=1.0, + ) + + def test_write_environments__compress_dynamo_documents_enabled__logs_expected( environment: Environment, dynamo_environment_wrapper: DynamoEnvironmentWrapper, @@ -120,6 +164,13 @@ def test_write_environments__compress_dynamo_documents_enabled__logs_expected( "event": "environment-document-compressed", "level": "info", }, + { + "environment__id": environment.id, + "feature_states__count": 0, + "document__bytes": ANY, + "event": "environment_document.written", + "level": "info", + }, ] diff --git a/api/tests/unit/environments/dynamodb/wrappers/test_unit_dynamodb_environment_v2_wrapper.py b/api/tests/unit/environments/dynamodb/wrappers/test_unit_dynamodb_environment_v2_wrapper.py index bbf9cbff52aa..5ca6ea990047 100644 --- a/api/tests/unit/environments/dynamodb/wrappers/test_unit_dynamodb_environment_v2_wrapper.py +++ b/api/tests/unit/environments/dynamodb/wrappers/test_unit_dynamodb_environment_v2_wrapper.py @@ -1,5 +1,6 @@ import uuid from typing import Any +from unittest.mock import ANY from boto3.dynamodb.types import Binary from common.test_tools import AssertMetricFixture @@ -317,6 +318,13 @@ def test_environment_v2_wrapper__write_environments_with_compression__logs_expec "event": "environment-document-compressed", "level": "info", }, + { + "environment__id": environment.id, + "feature_states__count": 0, + "document__bytes": ANY, + "event": "environment_document.written", + "level": "info", + }, ] diff --git a/api/tests/unit/environments/test_unit_environments_models.py b/api/tests/unit/environments/test_unit_environments_models.py index cbba88ab635e..87754221e149 100644 --- a/api/tests/unit/environments/test_unit_environments_models.py +++ b/api/tests/unit/environments/test_unit_environments_models.py @@ -82,6 +82,33 @@ def test_environment_save__feature_default_enabled_changed__preserves_feature_st assert FeatureState.objects.count() == 1 +def test_environment_create__seeding_feature_states__is_marked_as_creating( + project: Project, + mocker: MockerFixture, +) -> None: + # Given + is_creating_while_seeding: list[bool] = [] + + def _record_is_creating(environment: Environment) -> None: + is_creating_while_seeding.append( + Environment.objects.filter(id=environment.id, is_creating=True).exists() + ) + + mocker.patch.object( + FeatureState, + "create_initial_feature_states_for_environment", + side_effect=_record_is_creating, + ) + + # When + environment = Environment.objects.create(name="Test environment", project=project) + + # Then + assert is_creating_while_seeding == [True] + assert environment.is_creating is False + assert Environment.objects.get(id=environment.id).is_creating is False + + def test_environment_clone__default__does_not_modify_original_instance( environment: Environment, ) -> None: diff --git a/api/tests/unit/environments/test_unit_environments_tasks.py b/api/tests/unit/environments/test_unit_environments_tasks.py index bba3ff57ac3f..3a210cbb24df 100644 --- a/api/tests/unit/environments/test_unit_environments_tasks.py +++ b/api/tests/unit/environments/test_unit_environments_tasks.py @@ -1,4 +1,11 @@ +from datetime import timedelta + +from django.utils import timezone +from freezegun import freeze_time +from pytest_django.fixtures import SettingsWrapper from pytest_mock import MockerFixture +from pytest_structlog import StructuredLogCapture +from task_processor.task_run_method import TaskRunMethod from audit.models import AuditLog from environments.models import Environment @@ -7,6 +14,7 @@ process_environment_update, rebuild_environment_document, ) +from features.models import Feature def test_rebuild_environment_document__valid_environment__calls_write_documents( @@ -89,6 +97,125 @@ def test_process_environment_update__project_audit_log__sends_project_message( ) +def test_process_environment_update__environment_is_creating__defers_write( + environment: Environment, + settings: SettingsWrapper, + mocker: MockerFixture, +) -> None: + # Given + settings.TASK_RUN_METHOD = TaskRunMethod.TASK_PROCESSOR + Environment.objects.filter(id=environment.id).update(is_creating=True) + audit_log = AuditLog.objects.create( + project=environment.project, environment=environment + ) + mock_write_environment_documents = mocker.patch( + "environments.tasks.Environment.write_environment_documents", + ) + mock_task = mocker.patch("environments.tasks.process_environment_update") + + # When + with freeze_time("2099-01-01T00:00:00Z"): + process_environment_update(audit_log_id=audit_log.id) + expected_delay_until = timezone.now() + timedelta(seconds=1) + + # Then + mock_write_environment_documents.assert_not_called() + mock_task.delay.assert_called_once_with( + kwargs={"audit_log_id": audit_log.id, "deferrals": 1}, + delay_until=expected_delay_until, + ) + + +def test_process_environment_update__feature_is_creating__defers_write( + environment: Environment, + feature: Feature, + settings: SettingsWrapper, + mocker: MockerFixture, +) -> None: + # Given + settings.TASK_RUN_METHOD = TaskRunMethod.TASK_PROCESSOR + Feature.objects.filter(id=feature.id).update(is_creating=True) + audit_log = AuditLog.objects.create( + project=environment.project, environment=environment + ) + mock_write_environment_documents = mocker.patch( + "environments.tasks.Environment.write_environment_documents", + ) + mock_task = mocker.patch("environments.tasks.process_environment_update") + + # When + process_environment_update(audit_log_id=audit_log.id) + + # Then + mock_write_environment_documents.assert_not_called() + mock_task.delay.assert_called_once() + + +def test_process_environment_update__deferrals_exhausted__writes_document_and_warns( + environment: Environment, + settings: SettingsWrapper, + mocker: MockerFixture, + log: StructuredLogCapture, +) -> None: + # Given + settings.TASK_RUN_METHOD = TaskRunMethod.TASK_PROCESSOR + Environment.objects.filter(id=environment.id).update(is_creating=True) + audit_log = AuditLog.objects.create( + project=environment.project, environment=environment + ) + mock_write_environment_documents = mocker.patch( + "environments.tasks.Environment.write_environment_documents", + ) + mock_task = mocker.patch("environments.tasks.process_environment_update") + + # When + process_environment_update(audit_log_id=audit_log.id, deferrals=3) + + # Then + mock_task.delay.assert_not_called() + mock_write_environment_documents.assert_called_once_with( + environment_id=environment.id, project_id=environment.project.id + ) + assert log.has( + "environment_document.written_while_seeding", + level="warning", + audit_log__id=audit_log.id, + environment__id=environment.id, + project__id=environment.project.id, + ) + + +def test_process_environment_update__no_task_processor__writes_document( + environment: Environment, + settings: SettingsWrapper, + mocker: MockerFixture, +) -> None: + # Given + # `delay_until` is a no-op outside the task processor, so deferring would drop + # the write instead of retrying it. + settings.TASK_RUN_METHOD = TaskRunMethod.SEPARATE_THREAD + mock_is_seeding = mocker.patch("environments.tasks._is_seeding_feature_states") + Environment.objects.filter(id=environment.id).update(is_creating=True) + audit_log = AuditLog.objects.create( + project=environment.project, environment=environment + ) + mock_write_environment_documents = mocker.patch( + "environments.tasks.Environment.write_environment_documents", + ) + mock_task = mocker.patch("environments.tasks.process_environment_update") + + # When + process_environment_update(audit_log_id=audit_log.id) + + # Then + mock_task.delay.assert_not_called() + mock_write_environment_documents.assert_called_once_with( + environment_id=environment.id, project_id=environment.project.id + ) + # The `is_creating` check is skipped entirely, since it could not be acted on. + mock_is_seeding.assert_not_called() + + def test_delete_environment_from_dynamo__valid_environment__calls_all_wrappers( mocker: MockerFixture, ) -> None: diff --git a/api/tests/unit/features/test_unit_features_models.py b/api/tests/unit/features/test_unit_features_models.py index f81be84a3f97..250731462f4a 100644 --- a/api/tests/unit/features/test_unit_features_models.py +++ b/api/tests/unit/features/test_unit_features_models.py @@ -35,6 +35,34 @@ tomorrow = now + timedelta(days=1) +def test_feature_create__seeding_feature_states__is_marked_as_creating( + db: None, + project: Project, + mocker: MockerFixture, +) -> None: + # Given + is_creating_while_seeding: list[bool] = [] + + def _record_is_creating(feature: Feature) -> None: + is_creating_while_seeding.append( + Feature.objects.filter(id=feature.id, is_creating=True).exists() + ) + + mocker.patch.object( + FeatureState, + "create_initial_feature_states_for_feature", + side_effect=_record_is_creating, + ) + + # When + feature = Feature.objects.create(name="test_feature", project=project) + + # Then + assert is_creating_while_seeding == [True] + assert feature.is_creating is False + assert Feature.objects.get(id=feature.id).is_creating is False + + def test_feature_create__multiple_environments__creates_feature_states_for_all( db: None, environment: Environment, diff --git a/api/util/dataclasses.py b/api/util/dataclasses.py index f58ad3dd9bbe..07d431f2df9b 100644 --- a/api/util/dataclasses.py +++ b/api/util/dataclasses.py @@ -10,3 +10,4 @@ class CompressedEnvironmentDocument: document: Document compressed_size_bytes: int compression_ratio: float + feature_states_count: int diff --git a/api/util/mappers/dynamodb.py b/api/util/mappers/dynamodb.py index f59458772629..7f1048dc143f 100644 --- a/api/util/mappers/dynamodb.py +++ b/api/util/mappers/dynamodb.py @@ -200,6 +200,7 @@ def _get_compressed_environment_document( adapter: "TypeAdapter[Any]", ) -> CompressedEnvironmentDocument: uncompressed_size_bytes = estimate_document_size(document) + feature_states_count = len(cast(List[DocumentValue], document["feature_states"])) document["compressed"] = True compressed_document = adapter.validate_python(document) compressed_size_bytes = estimate_document_size(compressed_document) @@ -207,6 +208,7 @@ def _get_compressed_environment_document( document=cast(Document, compressed_document), compressed_size_bytes=compressed_size_bytes, compression_ratio=compressed_size_bytes / uncompressed_size_bytes, + feature_states_count=feature_states_count, ) diff --git a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md index 591712fced24..0bb34d58b6e4 100644 --- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md @@ -103,20 +103,40 @@ Attributes: ### `core.encrypted_field.decrypt_failed` Logged at `warning` from: - - `api/core/fields.py:37` + - `api/core/fields.py:62` Attributes: - `exc_info` +### `dynamodb.environment_document.written` + +Logged at `info` from: + - `api/environments/dynamodb/wrappers/environment_wrapper.py:114` + +Attributes: + - `document.bytes` + - `environment.id` + - `feature_states.count` + ### `dynamodb.environment_document_compressed` Logged at `info` from: - - `api/environments/dynamodb/wrappers/environment_wrapper.py:92` + - `api/environments/dynamodb/wrappers/environment_wrapper.py:95` Attributes: - `environment_api_key` - `environment_id` +### `environments.environment_document.written_while_seeding` + +Logged at `warning` from: + - `api/environments/tasks.py:83` + +Attributes: + - `audit_log.id` + - `environment.id` + - `project.id` + ### `experimentation.exposures.compute_failed` Logged at `error` from: diff --git a/docs/docs/deployment-self-hosting/observability/_metrics-catalogue.md b/docs/docs/deployment-self-hosting/observability/_metrics-catalogue.md index 203a40345f6d..f9a4386e9bb5 100644 --- a/docs/docs/deployment-self-hosting/observability/_metrics-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_metrics-catalogue.md @@ -43,6 +43,15 @@ Counter. Results of cache retrieval for environment document. `result` label is either `hit` or `miss`. +Labels: + - `result` + +### `flagsmith_environment_document_writes` + +Counter. + +Environment documents written to DynamoDB. `result` label is either `success` or `failure`. + Labels: - `result` From 2b4ddd7f5c5163116d271dbd76aeb5ba712781e2 Mon Sep 17 00:00:00 2001 From: Deep Santoshwar Date: Sat, 8 Aug 2026 17:09:51 +0530 Subject: [PATCH 2/2] fix(environments): report environment document writes after the batch is flushed `batch_writer` only buffers documents, flushing them when it fills up and when the context exits, so `put_item` returning says nothing about whether a document was written. Since these writes carry a single environment, or a project's handful, nothing was flushed before the success event was logged. Hold the events until the writer's context has exited, and count the documents a failed batch carried rather than only those already handed over. A failed batch does not report which of its documents were persisted, so the counter's description now says the failure count includes them. --- .../dynamodb/wrappers/environment_wrapper.py | 30 ++++++++++++------- api/environments/metrics.py | 3 +- .../test_unit_dynamo_environment_wrapper.py | 28 +++++++++++++++++ .../observability/_events-catalogue.md | 4 +-- .../observability/_metrics-catalogue.md | 2 +- 5 files changed, 52 insertions(+), 15 deletions(-) diff --git a/api/environments/dynamodb/wrappers/environment_wrapper.py b/api/environments/dynamodb/wrappers/environment_wrapper.py index e1bbf86a2e52..4a8d65577a56 100644 --- a/api/environments/dynamodb/wrappers/environment_wrapper.py +++ b/api/environments/dynamodb/wrappers/environment_wrapper.py @@ -72,10 +72,15 @@ def _write_environments(self, environments: Iterable["Environment"]) -> None: ) assert self.table - written = 0 + # `batch_writer` only buffers documents, flushing them when it fills up and + # when the context exits, so nothing is known to be written until then. + attempted = 0 + # (environment id, feature states count, document size in bytes) + pending_events: list[tuple[int, int, int]] = [] try: with self.table.batch_writer() as writer: for environment in environments: + attempted += 1 organisation = environment.project.organisation if openfeature_client.get_boolean_value( "compress_dynamo_documents", @@ -110,23 +115,26 @@ def _write_environments(self, environments: Iterable["Environment"]) -> None: compressed="false", ).observe(document_bytes) - written += 1 - logger.info( - "environment_document.written", - environment__id=environment.id, - feature_states__count=feature_states_count, - document__bytes=document_bytes, + pending_events.append( + (environment.id, feature_states_count, document_bytes) ) except Exception: - # The batch writer flushes on exit, so a failure can cover documents - # already handed to it as well as the one being written. + # A failed batch does not report which of its documents were persisted, + # so every document it carried is counted. flagsmith_environment_document_writes_total.labels(result="failure").inc( - max(written, 1) + attempted ) raise + for environment_id, feature_states_count, document_bytes in pending_events: + logger.info( + "environment_document.written", + environment__id=environment_id, + feature_states__count=feature_states_count, + document__bytes=document_bytes, + ) flagsmith_environment_document_writes_total.labels(result="success").inc( - written + attempted ) diff --git a/api/environments/metrics.py b/api/environments/metrics.py index 44a592e03f42..818ae183f784 100644 --- a/api/environments/metrics.py +++ b/api/environments/metrics.py @@ -17,7 +17,8 @@ flagsmith_environment_document_writes_total = prometheus_client.Counter( "flagsmith_environment_document_writes_total", "Environment documents written to DynamoDB. `result` label is either `success` or " - "`failure`.", + "`failure`. A batch write that fails counts every document it carried, since " + "DynamoDB does not report which of them were persisted.", ["result"], ) diff --git a/api/tests/unit/environments/dynamodb/wrappers/test_unit_dynamo_environment_wrapper.py b/api/tests/unit/environments/dynamodb/wrappers/test_unit_dynamo_environment_wrapper.py index 84375c3ae009..1c6ab76eb2f9 100644 --- a/api/tests/unit/environments/dynamodb/wrappers/test_unit_dynamo_environment_wrapper.py +++ b/api/tests/unit/environments/dynamodb/wrappers/test_unit_dynamo_environment_wrapper.py @@ -143,6 +143,34 @@ def test_write_environments__failed_write__counts_failure( ) +def test_write_environments__batch_flush_fails__counts_failure_without_logging( + environment: Environment, + assert_metric: AssertMetricFixture, + log: StructuredLogCapture, + mocker: MockerFixture, +) -> None: + # Given + # `put_item` only buffers the document — the batch is flushed when the writer's + # context exits, which is where a write can still fail. + dynamo_environment_wrapper = DynamoEnvironmentWrapper() + mocked_dynamo_table = mocker.patch.object(dynamo_environment_wrapper, "_table") + mocked_dynamo_table.batch_writer.return_value.__exit__.side_effect = RuntimeError( + "DynamoDB unavailable" + ) + + # When + with pytest.raises(RuntimeError): + dynamo_environment_wrapper.write_environments([environment]) + + # Then + assert not log.has("environment_document.written") + assert_metric( + name="flagsmith_environment_document_writes_total", + labels={"result": "failure"}, + value=1.0, + ) + + def test_write_environments__compress_dynamo_documents_enabled__logs_expected( environment: Environment, dynamo_environment_wrapper: DynamoEnvironmentWrapper, diff --git a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md index 0bb34d58b6e4..77e5f1f48667 100644 --- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md @@ -111,7 +111,7 @@ Attributes: ### `dynamodb.environment_document.written` Logged at `info` from: - - `api/environments/dynamodb/wrappers/environment_wrapper.py:114` + - `api/environments/dynamodb/wrappers/environment_wrapper.py:130` Attributes: - `document.bytes` @@ -121,7 +121,7 @@ Attributes: ### `dynamodb.environment_document_compressed` Logged at `info` from: - - `api/environments/dynamodb/wrappers/environment_wrapper.py:95` + - `api/environments/dynamodb/wrappers/environment_wrapper.py:100` Attributes: - `environment_api_key` diff --git a/docs/docs/deployment-self-hosting/observability/_metrics-catalogue.md b/docs/docs/deployment-self-hosting/observability/_metrics-catalogue.md index f9a4386e9bb5..2dc453fbf84b 100644 --- a/docs/docs/deployment-self-hosting/observability/_metrics-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_metrics-catalogue.md @@ -50,7 +50,7 @@ Labels: Counter. -Environment documents written to DynamoDB. `result` label is either `success` or `failure`. +Environment documents written to DynamoDB. `result` label is either `success` or `failure`. A batch write that fails counts every document it carried, since DynamoDB does not report which of them were persisted. Labels: - `result`