Perform garbage collection per RTO10

Have not yet implemented RTO10b2's server-specified grace period; have
split this work into #32.

TODO note based on spec at 488e932

Author: lawrence-forooghian

Sources/AblyLiveObjects/Internal/InternalDefaultLiveCounter.swift

@@ -242,6 +242,13 @@ internal final class InternalDefaultLiveCounter: Sendable {
         }
     }
 
+    /// Returns the object's RTLO3e `tombstonedAt` property.
+    internal var tombstonedAt: Date? {
+        mutex.withLock {
+            mutableState.liveObjectMutableState.tombstonedAt
+        }
+    }
+
     // MARK: - Mutable state and the operations that affect it
 
     private struct MutableState: InternalLiveObject {

Sources/AblyLiveObjects/Internal/InternalDefaultLiveMap.swift

@@ -392,6 +392,13 @@ internal final class InternalDefaultLiveMap: Sendable {
         }
     }
 
+    /// Releases entries that were tombstoned more than `gracePeriod` ago, per RTLM19.
+    internal func releaseTombstonedEntries(gracePeriod: TimeInterval, clock: SimpleClock) {
+        mutex.withLock {
+            mutableState.releaseTombstonedEntries(gracePeriod: gracePeriod, logger: logger, clock: clock)
+        }
+    }
+
     // MARK: - LiveObject
 
     /// Returns the object's RTLO3d `isTombstone` property.
@@ -401,6 +408,13 @@ internal final class InternalDefaultLiveMap: Sendable {
         }
     }
 
+    /// Returns the object's RTLO3e `tombstonedAt` property.
+    internal var tombstonedAt: Date? {
+        mutex.withLock {
+            mutableState.liveObjectMutableState.tombstonedAt
+        }
+    }
+
     // MARK: - Mutable state and the operations that affect it
 
     private struct MutableState: InternalLiveObject {
@@ -806,6 +820,29 @@ internal final class InternalDefaultLiveMap: Sendable {
             // RTLM4
             data = [:]
         }
+
+        /// Releases entries that were tombstoned more than `gracePeriod` ago, per RTLM19.
+        internal mutating func releaseTombstonedEntries(
+            gracePeriod: TimeInterval,
+            logger: Logger,
+            clock: SimpleClock,
+        ) {
+            let now = clock.now
+
+            // RTLM19a, RTLM19a1
+            data = data.filter { key, entry in
+                let shouldRelease = {
+                    guard let tombstonedAt = entry.tombstonedAt else {
+                        return false
+                    }
+
+                    return now.timeIntervalSince(tombstonedAt) >= gracePeriod
+                }()
+
+                logger.log("Releasing tombstoned entry \(entry) for key \(key)", level: .debug)
+                return !shouldRelease
+            }
+        }
     }
 
     // MARK: - Helper Methods

Sources/AblyLiveObjects/Internal/InternalDefaultRealtimeObjects.swift

@@ -18,6 +18,26 @@ internal final class InternalDefaultRealtimeObjects: Sendable, LiveMapObjectPool
     private let receivedObjectSyncProtocolMessages: AsyncStream<[InboundObjectMessage]>
     private let receivedObjectSyncProtocolMessagesContinuation: AsyncStream<[InboundObjectMessage]>.Continuation
 
+    /// The RTO10a interval at which we will perform garbage collection.
+    private let garbageCollectionInterval: TimeInterval
+    /// The RTO10b grace period for which we will retain tombstoned objects and map entries.
+    private nonisolated(unsafe) var garbageCollectionGracePeriod: TimeInterval
+    // The task that runs the periodic garbage collection described in RTO10.
+    private nonisolated(unsafe) var garbageCollectionTask: Task<Void, Never>!
+
+    /// Parameters used to control the garbage collection of tombstoned objects and map entries, as described in RTO10.
+    internal struct GarbageCollectionOptions {
+        /// The RTO10a interval at which we will perform garbage collection.
+        ///
+        /// The default value comes from the suggestion in RTO10a.
+        internal var interval: TimeInterval = 5 * 60
+
+        /// The initial RTO10b grace period for which we will retain tombstoned objects and map entries. This value may later get overridden by the `gcGracePeriod` of a `CONNECTED` `ProtocolMessage` from Realtime.
+        ///
+        /// This default value comes from RTO10b3; can be overridden for testing.
+        internal var gracePeriod: TimeInterval = 24 * 60 * 60
+    }
+
     internal var testsOnly_objectsPool: ObjectsPool {
         mutex.withLock {
             mutableState.objectsPool
@@ -71,14 +91,38 @@ internal final class InternalDefaultRealtimeObjects: Sendable, LiveMapObjectPool
         }
     }
 
-    internal init(logger: AblyPlugin.Logger, userCallbackQueue: DispatchQueue, clock: SimpleClock) {
+    internal init(logger: AblyPlugin.Logger, userCallbackQueue: DispatchQueue, clock: SimpleClock, garbageCollectionOptions: GarbageCollectionOptions = .init()) {
         self.logger = logger
         self.userCallbackQueue = userCallbackQueue
         self.clock = clock
         (receivedObjectProtocolMessages, receivedObjectProtocolMessagesContinuation) = AsyncStream.makeStream()
         (receivedObjectSyncProtocolMessages, receivedObjectSyncProtocolMessagesContinuation) = AsyncStream.makeStream()
         (waitingForSyncEvents, waitingForSyncEventsContinuation) = AsyncStream.makeStream()
         mutableState = .init(objectsPool: .init(logger: logger, userCallbackQueue: userCallbackQueue, clock: clock))
+        garbageCollectionInterval = garbageCollectionOptions.interval
+        garbageCollectionGracePeriod = garbageCollectionOptions.gracePeriod
+
+        garbageCollectionTask = Task { [weak self, garbageCollectionInterval] in
+            do {
+                while true {
+                    logger.log("Will perform garbage collection in \(garbageCollectionInterval)s", level: .debug)
+                    try await Task.sleep(nanoseconds: UInt64(garbageCollectionInterval) * NSEC_PER_SEC)
+
+                    guard let self else {
+                        return
+                    }
+
+                    performGarbageCollection()
+                }
+            } catch {
+                precondition(error is CancellationError)
+                logger.log("Garbage collection task terminated due to cancellation", level: .debug)
+            }
+        }
+    }
+
+    deinit {
+        garbageCollectionTask.cancel()
     }
 
     // MARK: - LiveMapObjectPoolDelegate
@@ -297,6 +341,19 @@ internal final class InternalDefaultRealtimeObjects: Sendable, LiveMapObjectPool
         try await coreSDK.publish(objectMessages: objectMessages)
     }
 
+    // MARK: - Garbage collection of deleted objects and map entries
+
+    /// Performs garbage collection of tombstoned objects and map entries, per RTO10c.
+    internal func performGarbageCollection() {
+        mutex.withLock {
+            mutableState.objectsPool.performGarbageCollection(
+                gracePeriod: garbageCollectionGracePeriod,
+                clock: clock,
+                logger: logger,
+            )
+        }
+    }
+
     // MARK: - Testing
 
     /// Finishes the following streams, to allow a test to perform assertions about which elements the streams have emitted to this moment:

Sources/AblyLiveObjects/Internal/ObjectsPool.swift

@@ -114,6 +114,15 @@ internal struct ObjectsPool {
                 map.isTombstone
             }
         }
+
+        internal var tombstonedAt: Date? {
+            switch self {
+            case let .counter(counter):
+                counter.tombstonedAt
+            case let .map(map):
+                map.tombstonedAt
+            }
+        }
     }
 
     /// Keyed by `objectId`.
@@ -399,4 +408,30 @@ internal struct ObjectsPool {
         // TODO: this one is unclear (are we meant to replace the root or just clear its data?) https://github.com/ably/specification/pull/333/files#r2183493458. I believe that the answer is that we should just clear its data but the spec point needs to be clearer, see https://github.com/ably/specification/pull/346/files#r2201434895.
         root.resetData()
     }
+
+    /// Performs garbage collection of tombstoned objects and map entries, per RTO10c.
+    internal mutating func performGarbageCollection(gracePeriod: TimeInterval, clock: SimpleClock, logger: Logger) {
+        logger.log("Performing garbage collection, grace period \(gracePeriod)s", level: .debug)
+
+        let now = clock.now
+
+        entries = entries.filter { key, entry in
+            if case let .map(map) = entry {
+                // RTO10c1a
+                map.releaseTombstonedEntries(gracePeriod: gracePeriod, clock: clock)
+            }
+
+            // RTO10c1b
+            let shouldRelease = {
+                guard let tombstonedAt = entry.tombstonedAt else {
+                    return false
+                }
+
+                return now.timeIntervalSince(tombstonedAt) >= gracePeriod
+            }()
+
+            logger.log("Releasing tombstoned entry \(entry) for key \(key)", level: .debug)
+            return !shouldRelease
+        }
+    }
 }