Coverage for object_streams/retention.py: 94%

1"""Retention settings and pruning for the replayable outbox.""" 

2 

3from __future__ import annotations 

4 

5from datetime import datetime 

6from datetime import timedelta 

7 

8from django.conf import settings 

9from django.db import transaction 

10from django.utils import timezone 

11 

12from object_streams.models import ObjectStreamEvent 

13from object_streams.outbox import broadcasted_through_cursor 

14from object_streams.outbox import record_pruned_through 

15 

16 

17__all__ = ( 

18 "get_retention_days", 

19 "get_retention_max_rows", 

20 "prune_outbox", 

21 "retention_cutoff", 

22) 

23 

24 

25def _positive_int_setting(name: str) -> int | None: 

26 value = getattr(settings, name, None) 

27 if value is None: 

28 return None 

29 value = int(value) 

30 if value < 1: 

31 msg = f"{name} must be a positive integer or None." 

32 raise ValueError(msg) 

33 return value 

34 

35 

36def get_retention_days() -> int | None: 

37 """Return the configured outbox age limit in days, or None to keep every row.""" 

38 

39 return _positive_int_setting("OBJECT_STREAMS_RETENTION_DAYS") 

40 

41 

42def get_retention_max_rows() -> int | None: 

43 """Return the configured outbox row limit, or None to keep every row.""" 

44 

45 return _positive_int_setting("OBJECT_STREAMS_RETENTION_MAX_ROWS") 

46 

47 

48def retention_cutoff(days: int) -> datetime: 

49 """Return the timestamp before which rows are older than the age limit.""" 

50 

51 if days < 1: 

52 msg = "Retention days must be a positive integer." 

53 raise ValueError(msg) 

54 return timezone.now() - timedelta(days=days) 

55 

56 

57def _manager(using: str | None): 

58 manager = ObjectStreamEvent.objects 

59 if using is not None: 

60 manager = manager.db_manager(using) 

61 return manager 

62 

63 

64def _age_prune_through(manager, before: datetime) -> int | None: 

65 return manager.filter(created_at__lt=before).order_by("-cursor").values_list("cursor", flat=True).first() 

66 

67 

68def _row_limit_prune_through(manager, max_rows: int) -> int | None: 

69 if max_rows < 1: 69 ↛ 70line 69 didn't jump to line 70 because the condition on line 69 was never true

70 msg = "Retention row limits must be a positive integer." 

71 raise ValueError(msg) 

72 oldest_kept = list(manager.order_by("-cursor").values_list("cursor", flat=True)[max_rows - 1 : max_rows]) 

73 if not oldest_kept: 

74 return None 

75 return oldest_kept[0] - 1 

76 

77 

78def prune_outbox( 

79 *, 

80 before: datetime | None = None, 

81 max_rows: int | None = None, 

82 using: str | None = None, 

83 dry_run: bool = False, 

84) -> int: 

85 """Delete outbox rows past the retention limits and return how many were removed. 

86 

87 Pruning never deletes the newest retained row, so the global cursor never 

88 moves backwards. Deleted ranges are recorded as a watermark so replay can 

89 answer with a resync instead of an empty catch-up. 

90 """ 

91 

92 if before is None and max_rows is None: 

93 return 0 

94 if max_rows is not None and max_rows < 1: 

95 msg = "Retention row limits must be a positive integer." 

96 raise ValueError(msg) 

97 

98 manager = _manager(using).filter(cursor__isnull=False) 

99 newest = manager.order_by("-cursor").values_list("cursor", flat=True).first() 

100 if newest is None: 

101 return 0 

102 

103 candidates = [] 

104 if before is not None: 

105 candidates.append(_age_prune_through(manager, before)) 

106 if max_rows is not None: 

107 candidates.append(_row_limit_prune_through(manager, max_rows)) 

108 

109 retained = [candidate for candidate in candidates if candidate is not None] 

110 if not retained: 

111 return 0 

112 

113 prune_through = min(max(retained), newest - 1, broadcasted_through_cursor(using=using)) 

114 if prune_through < 1: 114 ↛ 115line 114 didn't jump to line 115 because the condition on line 114 was never true

115 return 0 

116 

117 queryset = manager.filter(cursor__lte=prune_through) 

118 if dry_run: 

119 return queryset.count() 

120 

121 with transaction.atomic(using=using): 

122 deleted, _ = queryset.delete() 

123 if deleted: 123 ↛ 125line 123 didn't jump to line 125

124 record_pruned_through(prune_through, using=using) 

125 return deleted