M8-R08: add atomic meter close and binding transfer

This commit is contained in:
2026-08-24 18:37:46 +02:00
parent 2be4f78f8a
commit 8dc3f71aaf
15 changed files with 1667 additions and 35 deletions
+83 -1
View File
@@ -16,6 +16,7 @@ from app.models.energy import DsmrReading, Meter
from app.models.meter_source import MeterSource, MeterSourceBinding, MeterSourceChannel, WarmteLinkReading
from app.schemas.meter_source import (
BindingCreate, BindingListResponse, BindingPatch, BindingResponse, ChannelBindingSummaryResponse,
BindingTransferRequest, BindingTransferResponse,
ChannelReadingResponse,
ChannelReadingsResponse, CommoditiesResponse, CommodityResponse, DiscoverResponse,
DiscoverChannelResponse,
@@ -29,8 +30,9 @@ from app.services.dsmr_ingest import apply_dsmr_subscription
from app.services.meter_sources import (
BindingNotFoundError, ChannelNotFoundError, MeterNotFoundError,
MeterSourceError, SourceDeleteRestrictedError, SourceNotFoundError, create_binding,
create_source, delete_source, list_bindings, list_sources, update_binding, update_source,
create_source, delete_source, list_bindings, list_sources, transfer_binding, update_binding, update_source,
)
from app.services.energy_cost import recompute_range as electricity_recompute_range
from app.services import timezone as _tz_mod
from app.services.warmtelink_worker import warmtelink_worker_manager
@@ -104,6 +106,25 @@ def _binding_error(exc: MeterSourceError) -> HTTPException:
return HTTPException(status_code=422, detail=str(exc))
def _recompute_binding_commodity(db: Session, commodity: str, start: datetime) -> None:
end = datetime.now(UTC)
if start >= end:
return
if commodity == "electricity":
electricity_recompute_range(db, start, end, commit=False, strict=True)
else:
from app.services.meter_cost import recompute_range
recompute_range(db, start, end, commit=False)
def _republish_after_commit(db: Session) -> None:
try:
from app.services.ha_discovery import publish_discovery
publish_discovery(db)
except Exception:
pass
@router.get("/source-profiles", response_model=SourceProfilesResponse)
def source_profiles(_auth: AuthenticatedSession = Depends(require_session)) -> SourceProfilesResponse:
"""Return profile metadata; default secrets are never populated with stored values."""
@@ -300,12 +321,19 @@ def post_meter_binding(meter_id: int, body: BindingCreate, db: Session = Depends
try:
binding = create_binding(db, meter_id=meter_id, channel_id=channel.id, started_at=_as_utc(body.started_at),
ended_at=_as_utc(body.ended_at) if body.ended_at else None)
meter = db.get(Meter, meter_id)
db.flush()
_recompute_binding_commodity(db, meter.commodity, _as_utc(body.started_at))
db.commit()
db.refresh(binding)
_republish_after_commit(db)
return binding_response(binding)
except MeterSourceError as exc:
db.rollback()
raise _binding_error(exc) from exc
except Exception:
db.rollback()
raise
@router.patch("/bindings/{binding_uuid}", response_model=BindingResponse)
@@ -323,10 +351,64 @@ def patch_binding(binding_uuid: str, body: BindingPatch, db: Session = Depends(g
changes["started_at"] = _as_utc(body.started_at) if body.started_at is not None else None
if "ended_at" in body.model_fields_set:
changes["ended_at"] = _as_utc(body.ended_at) if body.ended_at is not None else None
old_started_at = _as_utc(binding.started_at)
old_ended_at = _as_utc(binding.ended_at) if binding.ended_at is not None else None
updated = update_binding(db, binding.id, **changes)
meter = db.get(Meter, updated.meter_id)
if "started_at" in changes:
earliest = min(old_started_at, _as_utc(updated.started_at))
elif "ended_at" in changes:
new_ended_at = _as_utc(updated.ended_at) if updated.ended_at is not None else None
changed_ends = [value for value in (old_ended_at, new_ended_at) if value is not None]
earliest = min(changed_ends) if changed_ends else old_started_at
else:
earliest = old_started_at
db.flush()
_recompute_binding_commodity(db, meter.commodity, earliest)
db.commit()
db.refresh(updated)
_republish_after_commit(db)
return binding_response(updated)
except MeterSourceError as exc:
db.rollback()
raise _binding_error(exc) from exc
except Exception:
db.rollback()
raise
@router.post("/meters/{meter_id}/bindings/transfer", response_model=BindingTransferResponse)
def post_binding_transfer(
meter_id: int, body: BindingTransferRequest, db: Session = Depends(get_db),
_auth: AuthenticatedSession = Depends(require_session), _csrf: None = Depends(require_csrf),
) -> BindingTransferResponse:
source = db.execute(select(MeterSourceBinding).where(
MeterSourceBinding.uuid == body.from_binding_uuid
)).scalar_one_or_none()
channel = db.execute(select(MeterSourceChannel).where(
MeterSourceChannel.uuid == body.to_source_channel_uuid
)).scalar_one_or_none()
if source is None or channel is None:
raise HTTPException(status_code=404, detail="Meter source binding or channel not found.")
effective_at = _as_utc(body.effective_at)
try:
closed, created = transfer_binding(db, target_meter_id=meter_id, from_binding_id=source.id,
to_channel_id=channel.id, effective_at=effective_at)
meter = db.get(Meter, meter_id)
if closed.meter_id == meter.id:
earliest = effective_at
else:
earliest = min(_as_utc(closed.ended_at), effective_at)
db.flush()
_recompute_binding_commodity(db, meter.commodity, earliest)
db.commit()
db.refresh(closed)
db.refresh(created)
except MeterSourceError as exc:
db.rollback()
raise _binding_error(exc) from exc
except Exception:
db.rollback()
raise
_republish_after_commit(db)
return BindingTransferResponse(closed_binding=binding_response(closed), created_binding=binding_response(created))
+64 -4
View File
@@ -55,6 +55,7 @@ from app.dependencies import get_db
from app.models.energy import Meter
from app.models.meter_source import MeterSourceChannel
from app.schemas.meter import (
MeterCloseRequest,
MeterDeclareRequest,
MeterBindingSummary,
MeterListResponse,
@@ -66,6 +67,7 @@ from app.services.meter_sources import (
MeterSourceError,
create_binding,
create_binding_for_meter_swap,
close_open_bindings_for_meter,
)
from app.services import timezone as _tz_mod
from app.services.auth import AuthenticatedSession
@@ -74,6 +76,7 @@ from app.services.meters import (
MeterIntervalError,
MeterOverlapError,
declare_meter,
close_meter,
list_meters,
update_meter,
)
@@ -151,7 +154,7 @@ def _trigger_recompute(db: Session, start: datetime, label: str) -> int:
# started_at is in the future — nothing to recompute.
logger.info("%s: started_at (%s) is in the future, skipping recompute.", label, start)
return 0
n = recompute_range(db, start, end, commit=False)
n = recompute_range(db, start, end, commit=False, strict=True)
logger.info(
"%s: recomputed %d period(s) in window [%s, %s).",
label,
@@ -162,6 +165,17 @@ def _trigger_recompute(db: Session, start: datetime, label: str) -> int:
return n
def _recompute_commodity(db: Session, commodity: str, start: datetime, label: str) -> int:
if commodity == "electricity":
return _trigger_recompute(db, start, label)
from app.services.meter_cost import recompute_range as thermal_recompute_range
end = datetime.now(UTC)
if start >= end:
return 0
return thermal_recompute_range(db, start, end, commit=False)
def _meter_response(meter: Meter) -> MeterResponse:
"""Serialize meter plus binding summaries without exposing source config."""
response = MeterResponse.model_validate(meter)
@@ -248,6 +262,16 @@ def declare_energy_meter(
note=body.note,
)
db.flush() # assign PK before an optional binding and recompute
# A closed predecessor must never retain an open interval. For a
# meter swap with no selected channel we can safely hand off exactly
# one compatible open channel; ambiguity is fail-closed.
auto_channel = None
if body.source_channel_uuid is None and old_meter is not None and body.reason.value == "meter_swap":
candidates = [b for b in old_meter.source_bindings if b.ended_at is None and b.channel.unit == {"electricity": "kWh", "heating": "GJ", "hot_water": ""}.get(body.commodity)]
if len(candidates) > 1:
raise MeterSourceError("Meter swap has ambiguous open bindings; select a channel explicitly.")
if len(candidates) == 1:
auto_channel = candidates[0].channel
if body.source_channel_uuid is not None:
channel = db.execute(
select(MeterSourceChannel).where(MeterSourceChannel.uuid == body.source_channel_uuid)
@@ -269,14 +293,20 @@ def declare_energy_meter(
channel_id=channel.id,
started_at=started_at_utc,
)
elif auto_channel is not None:
create_binding_for_meter_swap(db, old_meter_id=old_meter.id, new_meter_id=new_meter.id,
channel_id=auto_channel.id, started_at=started_at_utc)
if old_meter is not None:
close_open_bindings_for_meter(db, old_meter.id, ended_at=started_at_utc)
# Keep recompute in this transaction: a failure must not leave a new
# meter, its predecessor, or either binding at a half-applied boundary.
now = datetime.now(UTC)
if started_at_utc < now:
_trigger_recompute(db, started_at_utc, "POST /api/energy/meters")
db.flush()
_recompute_commodity(db, body.commodity, started_at_utc, "POST /api/energy/meters")
db.commit()
except (MeterOverlapError, MeterSourceError) as exc:
except (MeterIntervalError, MeterOverlapError, MeterSourceError) as exc:
db.rollback()
raise HTTPException(
status_code=(status.HTTP_404_NOT_FOUND if isinstance(exc, ChannelNotFoundError)
@@ -304,6 +334,30 @@ def declare_energy_meter(
return _meter_response(new_meter)
@router.post("/meters/{meter_id}/close", response_model=MeterResponse)
def close_energy_meter(
meter_id: int, body: MeterCloseRequest, db: Session = Depends(get_db),
_auth: AuthenticatedSession = Depends(require_session), _csrf: None = Depends(require_csrf),
) -> MeterResponse:
meter = _get_meter_or_404(db, meter_id)
boundary = _localize_started_at(body.ended_at)
try:
close_meter(db, meter, ended_at=boundary)
close_open_bindings_for_meter(db, meter.id, ended_at=boundary)
db.flush()
_recompute_commodity(db, meter.commodity, boundary, f"POST /api/energy/meters/{meter_id}/close")
db.commit()
except (MeterIntervalError, MeterSourceError) as exc:
db.rollback()
raise HTTPException(status_code=422, detail=str(exc)) from exc
except Exception:
db.rollback()
raise
db.refresh(meter)
_trigger_discovery_republish(db)
return _meter_response(meter)
# ---------------------------------------------------------------------------
# PATCH /api/energy/meters/{id}
# ---------------------------------------------------------------------------
@@ -361,7 +415,13 @@ def patch_energy_meter(
# Window = [min(old, new), now) — covers all periods whose attribution
# may have changed due to the boundary shift in either direction.
window_start = min(old_started_at, new_started_at_utc)
_trigger_recompute(db, window_start, f"PATCH /api/energy/meters/{meter_id}")
db.flush()
_recompute_commodity(
db,
meter.commodity,
window_start,
f"PATCH /api/energy/meters/{meter_id}",
)
db.commit()
except MeterIntervalError as exc:
+6
View File
@@ -146,3 +146,9 @@ class MeterPatchRequest(BaseModel):
"Triggers billing recompute over the affected window."
),
)
class MeterCloseRequest(BaseModel):
"""Close the active meter epoch at an exclusive end boundary."""
ended_at: datetime
+11
View File
@@ -135,6 +135,12 @@ class BindingPatch(BaseModel):
ended_at: datetime | None = None
class BindingTransferRequest(BaseModel):
from_binding_uuid: str = Field(..., min_length=1, max_length=36)
to_source_channel_uuid: str = Field(..., min_length=1, max_length=36)
effective_at: datetime
class BindingResponse(BaseModel):
uuid: str
meter_id: int
@@ -149,3 +155,8 @@ class BindingResponse(BaseModel):
class BindingListResponse(BaseModel):
items: list[BindingResponse]
total: int
class BindingTransferResponse(BaseModel):
closed_binding: BindingResponse
created_binding: BindingResponse
+8 -1
View File
@@ -666,7 +666,7 @@ def compute_closed_periods(session: Session) -> int:
def recompute_range(
session: Session, start: datetime, end: datetime, *, commit: bool = True
session: Session, start: datetime, end: datetime, *, commit: bool = True, strict: bool = False
) -> int:
"""Recompute (overwrite) all 15-minute periods in ``[start, end)``.
@@ -693,6 +693,11 @@ def recompute_range(
When true (the default), commit after all periods have been processed.
Callers composing this recompute with other writes may pass false and
own the surrounding transaction themselves.
strict:
When true, propagate a failed period computation to the caller. This
is for lifecycle transactions which must roll back their meter/binding
mutation together with the cost recompute. The default remains
best-effort for existing background and standalone callers.
start:
Inclusive start datetime (floored to the nearest quarter-hour internally).
end:
@@ -723,6 +728,8 @@ def recompute_range(
if did_write:
written += 1
except Exception:
if strict:
raise
logger.exception(
"recompute_range: unexpected error for t0=%s — continuing.",
t0.isoformat(),
+109 -4
View File
@@ -250,7 +250,7 @@ def _validate_binding(
channel_id: int,
started_at: datetime,
ended_at: datetime | None,
excluding_id: int | None = None,
excluding_ids: set[int] | None = None,
) -> None:
meter = _get_meter(session, meter_id)
channel = get_channel(session, channel_id)
@@ -264,7 +264,19 @@ def _validate_binding(
)
if ended_at is not None and _as_utc(ended_at) <= _as_utc(started_at):
raise BindingValidationError("Binding ended_at must be strictly after started_at.")
if _as_utc(started_at) < _as_utc(meter.started_at):
raise BindingValidationError("Binding must not start before its meter epoch.")
if meter.ended_at is None:
if ended_at is not None:
# A historical binding on an active epoch is valid, but it must be
# wholly within that epoch (whose upper bound is open).
pass
else:
meter_end = _as_utc(meter.ended_at)
if ended_at is None or _as_utc(ended_at) > meter_end:
raise BindingValidationError("Closed meter bindings must end within the meter epoch.")
excluded = excluding_ids or set()
candidates = session.execute(
select(MeterSourceBinding).where(
or_(
@@ -274,7 +286,7 @@ def _validate_binding(
)
).scalars()
for existing in candidates:
if existing.id == excluding_id:
if existing.id in excluded:
continue
if half_open_intervals_overlap(
_as_utc(started_at),
@@ -295,6 +307,9 @@ def create_binding(
ended_at: datetime | None = None,
) -> MeterSourceBinding:
"""Create a compatible non-overlapping half-open source binding."""
now = _utc_now()
if _as_utc(started_at) > now or (ended_at is not None and _as_utc(ended_at) > now):
raise BindingValidationError("Binding boundaries must not be in the future.")
_validate_binding(
session,
meter_id=meter_id,
@@ -302,7 +317,6 @@ def create_binding(
started_at=started_at,
ended_at=ended_at,
)
now = _utc_now()
binding = MeterSourceBinding(
meter_id=meter_id,
channel_id=channel_id,
@@ -403,13 +417,16 @@ def update_binding(
new_channel_id = binding.channel_id if channel_id is None else channel_id
new_started_at = binding.started_at if started_at is None else started_at
new_ended_at = binding.ended_at if ended_at is _UNSET else ended_at
now = _utc_now()
if _as_utc(new_started_at) > now or (new_ended_at is not None and _as_utc(new_ended_at) > now):
raise BindingValidationError("Binding boundaries must not be in the future.")
_validate_binding(
session,
meter_id=new_meter_id,
channel_id=new_channel_id,
started_at=new_started_at,
ended_at=new_ended_at,
excluding_id=binding.id,
excluding_ids={binding.id},
)
binding.meter_id = new_meter_id
binding.channel_id = new_channel_id
@@ -422,3 +439,91 @@ def update_binding(
def close_binding(session: Session, binding_id: int, *, ended_at: datetime) -> MeterSourceBinding:
"""Close an existing binding at its exclusive end boundary."""
return update_binding(session, binding_id, ended_at=ended_at)
def close_open_bindings_for_meter(session: Session, meter_id: int, *, ended_at: datetime) -> list[MeterSourceBinding]:
"""Close every open binding on a meter at one shared boundary."""
bindings = list(session.execute(
select(MeterSourceBinding).where(
MeterSourceBinding.meter_id == meter_id, MeterSourceBinding.ended_at.is_(None)
)
).scalars())
for binding in bindings:
update_binding(session, binding.id, ended_at=ended_at)
return bindings
def transfer_binding(
session: Session, *, target_meter_id: int, from_binding_id: int, to_channel_id: int,
effective_at: datetime,
) -> tuple[MeterSourceBinding, MeterSourceBinding]:
"""Atomically close a binding and open its replacement on the target meter."""
source = session.get(MeterSourceBinding, from_binding_id)
if source is None:
raise BindingNotFoundError(f"Meter source binding {from_binding_id} was not found.")
target = _get_meter(session, target_meter_id)
old_meter = _get_meter(session, source.meter_id)
effective_at = _as_utc(effective_at)
now = _utc_now()
if effective_at > now:
raise BindingValidationError("Binding transfer effective_at must not be in the future.")
if old_meter.commodity != target.commodity:
raise BindingValidationError("Binding transfer meters must have the same commodity.")
if source.ended_at is not None:
raise BindingValidationError("Only an open binding can be transferred.")
if old_meter.id == target.id:
close_at = effective_at
else:
# Recovery is deliberately narrow: the source meter must be the one
# and only most-recent closed predecessor in this commodity's timeline.
# A manually closed meter may leave an intentional epoch gap before the
# target is declared, so adjacency is not required.
if old_meter.ended_at is None:
raise BindingValidationError("Source binding must belong to a closed predecessor meter.")
timeline = list(session.execute(
select(Meter).where(Meter.commodity == target.commodity)
).scalars())
predecessors = [
meter for meter in timeline
if meter.id != target.id
and meter.ended_at is not None
and _as_utc(meter.ended_at) <= _as_utc(target.started_at)
]
if not predecessors:
raise BindingValidationError("Source meter is not the unique immediately preceding meter.")
latest_end = max(_as_utc(meter.ended_at) for meter in predecessors)
latest = [meter for meter in predecessors if _as_utc(meter.ended_at) == latest_end]
if len(latest) != 1 or latest[0].id != old_meter.id:
raise BindingValidationError("Source meter is not the unique immediately preceding meter.")
# Reject any overlapping epoch around either endpoint. A separate
# meter inside the gap is already excluded by the predecessor check;
# one extending into either endpoint is an ambiguous timeline too.
for meter in timeline:
if meter.id in {old_meter.id, target.id}:
continue
meter_end = _as_utc(meter.ended_at) if meter.ended_at is not None else None
if (
half_open_intervals_overlap(
_as_utc(old_meter.started_at), _as_utc(old_meter.ended_at),
_as_utc(meter.started_at), meter_end,
)
or half_open_intervals_overlap(
_as_utc(target.started_at),
_as_utc(target.ended_at) if target.ended_at is not None else None,
_as_utc(meter.started_at), meter_end,
)
):
raise BindingValidationError("Source meter has an ambiguous commodity timeline.")
close_at = _as_utc(old_meter.ended_at)
if effective_at < _as_utc(target.started_at):
raise BindingValidationError("Transfer effective_at must be within the target meter epoch.")
if effective_at < _as_utc(source.started_at):
raise BindingValidationError("Transfer effective_at precedes the source binding.")
# Validate the target before mutating the old row, then close/create in one session.
_validate_binding(session, meter_id=target.id, channel_id=to_channel_id,
started_at=effective_at, ended_at=None,
excluding_ids={source.id})
update_binding(session, source.id, ended_at=close_at)
created = create_binding(session, meter_id=target.id, channel_id=to_channel_id,
started_at=effective_at)
return source, created
+76 -2
View File
@@ -53,6 +53,7 @@ from sqlalchemy import select
from sqlalchemy.orm import Session
from app.models.energy import Meter
from app.models.meter_source import MeterSourceBinding
logger = logging.getLogger(__name__)
@@ -104,6 +105,20 @@ class MeterIntervalError(MeterError):
"""
def close_meter(session: Session, meter: Meter, *, ended_at: datetime) -> Meter:
"""Close an active meter at a valid, non-future exclusive boundary."""
boundary = _as_utc(ended_at)
if meter.ended_at is not None:
raise MeterIntervalError("Only an active meter can be closed.")
if boundary <= _as_utc(meter.started_at):
raise MeterIntervalError("Meter ended_at must be strictly after started_at.")
if boundary > datetime.now(UTC):
raise MeterIntervalError("Meter ended_at must not be in the future.")
_validate_bindings_fit_meter_end(session, meter, boundary)
meter.ended_at = boundary
return meter
# ---------------------------------------------------------------------------
# Internal query helpers
# ---------------------------------------------------------------------------
@@ -121,6 +136,23 @@ def _active_meter(session: Session, commodity: str) -> Optional[Meter]:
).scalar_one_or_none()
def _validate_bindings_fit_meter_end(session: Session, meter: Meter, boundary: datetime) -> None:
"""Reject an epoch close that would put any retained binding out of bounds.
Closed binding history is immutable here. Open bindings may subsequently
be closed by the caller at the shared meter boundary, but only when that
produces a non-empty interval.
"""
for binding in session.execute(
select(MeterSourceBinding).where(MeterSourceBinding.meter_id == meter.id)
).scalars():
if binding.ended_at is None:
if _as_utc(binding.started_at) >= boundary:
raise MeterIntervalError("Open binding cannot be closed within the proposed meter epoch.")
elif _as_utc(binding.ended_at) > boundary:
raise MeterIntervalError("Closed binding extends beyond the proposed meter epoch.")
def _meter_before(session: Session, meter: Meter) -> Optional[Meter]:
"""Return the meter whose ``ended_at`` equals *meter*'s ``started_at``.
@@ -296,6 +328,8 @@ def declare_meter(
If *started_at* is strictly earlier than the current active meter's
``started_at`` (chronological backdate below the active epoch's start).
"""
if _as_utc(started_at) > datetime.now(UTC):
raise MeterIntervalError("Meter started_at must not be in the future.")
active = _active_meter(session, commodity)
if active is not None:
@@ -308,6 +342,9 @@ def declare_meter(
"Declare a started_at on or after the active meter's start to avoid "
"a chronologically inconsistent epoch ordering."
)
# Validate before changing the epoch: retained closed binding history
# must never be silently truncated by a later declaration.
_validate_bindings_fit_meter_end(session, active, _as_utc(started_at))
# Close the current active meter at the swap point (contiguous handoff).
active.ended_at = started_at
logger.info(
@@ -400,6 +437,9 @@ def update_meter(
invert).
b. It must be **strictly before** this meter's ``ended_at`` (if set),
so this meter's epoch remains non-empty.
c. Every binding on this meter and its affected predecessor must remain
wholly inside its proposed epoch. The service rejects the correction
rather than rewriting binding history.
Note: triggering a billing recompute (``recompute_range``) after a
retroactive ``started_at`` change is **out of scope** for this service
@@ -429,6 +469,14 @@ def update_meter(
If the new ``started_at`` would produce an invalid (empty or inverted)
epoch for this meter or the immediately preceding one.
"""
# Validate the proposed epoch boundary before touching *any* mutable
# field. PATCH accepts label/note together with started_at, so doing this
# first keeps an invalid future timestamp from leaking a partial in-session
# update before the API's rollback boundary is reached.
proposed_started_at = _as_utc(started_at) if started_at is not None else None
if proposed_started_at is not None and proposed_started_at > datetime.now(UTC):
raise MeterIntervalError("Meter started_at must not be in the future.")
if label is not None:
meter.label = label
logger.info("Updated meter id=%d label=%r", meter.id, label)
@@ -439,10 +487,11 @@ def update_meter(
if started_at is not None:
old_started_at = meter.started_at
assert proposed_started_at is not None
# --- Validate upper bound: new started_at must be < this meter's ended_at (if set).
if meter.ended_at is not None:
if _as_utc(started_at) >= _as_utc(meter.ended_at):
if proposed_started_at >= _as_utc(meter.ended_at):
raise MeterIntervalError(
f"New started_at ({started_at.isoformat()}) must be strictly before "
f"this meter's ended_at ({meter.ended_at.isoformat()}). "
@@ -454,12 +503,37 @@ def update_meter(
# --- Validate lower bound: new started_at must be strictly after prev's started_at.
if prev is not None:
if _as_utc(started_at) <= _as_utc(prev.started_at):
if proposed_started_at <= _as_utc(prev.started_at):
raise MeterIntervalError(
f"New started_at ({started_at.isoformat()}) must be strictly after "
f"the previous meter's started_at ({prev.started_at.isoformat()}). "
"Moving the boundary that far back would collapse the previous meter's epoch."
)
# A boundary correction changes both adjacent meter epochs. Fail closed
# rather than silently rewriting binding history: every existing binding
# must still fit in its proposed epoch before either Meter is mutated.
affected_meters = [
(meter, proposed_started_at, _as_utc(meter.ended_at) if meter.ended_at is not None else None)
]
if prev is not None:
affected_meters.append((prev, _as_utc(prev.started_at), proposed_started_at))
for affected_meter, proposed_start, proposed_end in affected_meters:
bindings = session.execute(
select(MeterSourceBinding).where(MeterSourceBinding.meter_id == affected_meter.id)
).scalars()
for binding in bindings:
if _as_utc(binding.started_at) < proposed_start:
raise MeterIntervalError(
f"Binding {binding.id} starts before meter {affected_meter.id}'s epoch."
)
if proposed_end is not None and (
binding.ended_at is None or _as_utc(binding.ended_at) > proposed_end
):
raise MeterIntervalError(
f"Binding {binding.id} would fall outside meter {affected_meter.id}'s epoch."
)
if prev is not None:
# Maintain continuity: update the previous meter's ended_at to match the new start.
prev.ended_at = started_at
logger.info(