diff --git a/server/polar/reducer_bucket/redis_store.py b/server/polar/reducer_bucket/redis_store.py index a2c8224bab2..84ccdd6bdda 100644 --- a/server/polar/reducer_bucket/redis_store.py +++ b/server/polar/reducer_bucket/redis_store.py @@ -2,6 +2,7 @@ import uuid from collections.abc import Mapping, Sequence from datetime import datetime, timedelta +from decimal import Decimal import structlog @@ -33,7 +34,7 @@ def get_reducer_bucket_key( async def add_to_buckets( - redis: Redis, contributions: Mapping[BucketId, Sequence[float]] + redis: Redis, contributions: Mapping[BucketId, Sequence[Decimal]] ) -> None: """Adds the values to their buckets, skipping unique aggregations.""" ttl = int(REDUCER_BUCKET_TTL.total_seconds()) @@ -47,14 +48,14 @@ async def add_to_buckets( case AggregationFunction.cnt: await pipe.incrby(key, len(values)) case AggregationFunction.sum: - await pipe.incrbyfloat(key, sum(values)) + await pipe.incrbyfloat(key, float(sum(values))) case AggregationFunction.avg: await pipe.hincrby(key, "count", len(values)) - await pipe.hincrbyfloat(key, "sum", sum(values)) + await pipe.hincrbyfloat(key, "sum", float(sum(values))) case AggregationFunction.min: - await pipe.zadd(key, {"min": min(values)}, lt=True) + await pipe.zadd(key, {"min": float(min(values))}, lt=True) case AggregationFunction.max: - await pipe.zadd(key, {"max": max(values)}, gt=True) + await pipe.zadd(key, {"max": float(max(values))}, gt=True) case AggregationFunction.unique: log.info( "Ignoring reducer bucket of unique aggregation", diff --git a/server/polar/reducer_bucket/repository.py b/server/polar/reducer_bucket/repository.py index d6400f3c9ce..685b1b4df05 100644 --- a/server/polar/reducer_bucket/repository.py +++ b/server/polar/reducer_bucket/repository.py @@ -155,9 +155,9 @@ async def add_contribution( external_customer_id: str | None, *, count: int, - total: float = 0, - minimum: float | None = None, - maximum: float | None = None, + total: Decimal = Decimal(0), + minimum: Decimal | None = None, + maximum: Decimal | None = None, ) -> None: """Adds to the customer identity's latest bucket, or to the next generation once it's sealed.""" diff --git a/server/polar/reducer_bucket/service.py b/server/polar/reducer_bucket/service.py index 07707092ef5..4c5bfcbfb6c 100644 --- a/server/polar/reducer_bucket/service.py +++ b/server/polar/reducer_bucket/service.py @@ -28,7 +28,7 @@ REDUCER_BUCKET_SIZE = timedelta(minutes=5) -type Contributions = dict[redis_store.BucketId, list[tuple[datetime, float]]] +type Contributions = dict[redis_store.BucketId, list[tuple[datetime, Decimal]]] def get_reducer_bucket_start(timestamp: datetime) -> datetime: @@ -127,7 +127,7 @@ async def add_to_closed_buckets( continue statistics = ( { - "total": sum(values), + "total": sum(values, Decimal(0)), "minimum": min(values), "maximum": max(values), } @@ -468,12 +468,12 @@ def _reduce_events_to_bucket_contributions( if not reducer.filter.matches(event): continue aggregation = reducer.aggregation - value = 1.0 + value = Decimal(1) if isinstance(aggregation, PropertyAggregation): property_value = aggregation.get_value(event) if property_value is None: continue - value = property_value + value = Decimal(str(property_value)) bucket_id = ( reducer, get_reducer_bucket_start(event.timestamp), diff --git a/server/tests/reducer_bucket/test_service.py b/server/tests/reducer_bucket/test_service.py index 0a064078aa1..db9c205b9d4 100644 --- a/server/tests/reducer_bucket/test_service.py +++ b/server/tests/reducer_bucket/test_service.py @@ -134,7 +134,7 @@ async def test_adds_to_closed_buckets( organization: Organization, ) -> None: previous_start = BUCKET_START - REDUCER_BUCKET_SIZE - for batch in ((3, 4), (2,)): + for batch in ((0.1, 0.2), (0.3,)): await reducer_bucket_service.rollup( session, redis, @@ -150,8 +150,8 @@ async def test_adds_to_closed_buckets( ) [bucket] = await get_buckets(session) - assert (bucket.generation, bucket.count, bucket.sum) == (1, 3, Decimal(9)) - assert (bucket.min, bucket.max) == (Decimal(2), Decimal(4)) + assert (bucket.generation, bucket.count, bucket.sum) == (1, 3, Decimal("0.6")) + assert (bucket.min, bucket.max) == (Decimal("0.1"), Decimal("0.3")) async def test_adds_float_values_to_open_bucket( self, @@ -238,17 +238,17 @@ async def test_skips_organizations_without_syncable_reducers( ( PropertyAggregation(func=AggregationFunction.sum, property="tokens"), lambda redis, key: redis.get(key), - b"17", + b"18", ), ( PropertyAggregation(func=AggregationFunction.avg, property="tokens"), lambda redis, key: redis.hmget(key, ["count", "sum"]), - [b"6", b"17"], + [b"6", b"18"], ), ( PropertyAggregation(func=AggregationFunction.min, property="tokens"), lambda redis, key: redis.zscore(key, "min"), - 1.0, + 1.5, ), ( PropertyAggregation(func=AggregationFunction.max, property="tokens"), @@ -275,7 +275,7 @@ async def test_accumulates_across_calls( save_fixture, organization=organization, meters=[meter] ) - for batch in ((1, 2, 3), (5, 4, 2)): + for batch in ((1.5, 2, 3), (5, 4.5, 2)): events = [ build_event(organization, metadata={"tokens": tokens}) for tokens in batch