Make high priority events force an upload via upload conditions (#2699)

* Make high priority events force an upload via upload conditions

Forced uploads should be defined as an additional upload condition, as opposed to constructing an ad-hoc upload package. This allows the prioritizer and uploader to do housekeeping more easily.

Address a race condition during high-priority event sending in which a package can be asked for whilst the storage is still asking the prioritizer to deprioritize events that have already been removed from disk.

* Make _timeMillis in GDTClock be the actual unix time

* Style

Author: mikehaney24

GoogleDataTransport/GoogleDataTransport/Classes/GDTClock.m

@@ -81,7 +81,8 @@ - (instancetype)init {
   if (self) {
     _kernelBootTime = KernelBootTimeInNanoseconds();
     _uptime = UptimeInNanoseconds();
-    _timeMillis = CFAbsoluteTimeGetCurrent() * NSEC_PER_USEC;
+    _timeMillis =
+        (int64_t)((CFAbsoluteTimeGetCurrent() + kCFAbsoluteTimeIntervalSince1970) * NSEC_PER_USEC);
     CFTimeZoneRef timeZoneRef = CFTimeZoneCopySystem();
     _timezoneOffsetSeconds = CFTimeZoneGetSecondsFromGMT(timeZoneRef, 0);
     CFRelease(timeZoneRef);

GoogleDataTransport/GoogleDataTransport/Classes/GDTStorage.m

@@ -83,42 +83,37 @@ - (void)storeEvent:(GDTEvent *)event {
     // Add event to tracking collections.
     [self addEventToTrackingCollections:storedEvent];
 
+    // Have the prioritizer prioritize the event.
+    [prioritizer prioritizeEvent:storedEvent];
+
     // Check the QoS, if it's high priority, notify the target that it has a high priority event.
     if (event.qosTier == GDTEventQoSFast) {
-      NSSet<GDTStoredEvent *> *allEventsForTarget = self.targetToEventSet[storedEvent.target];
-      [self.uploader forceUploadEvents:allEventsForTarget target:target];
+      [self.uploader forceUploadForTarget:target];
     }
-
-    // Have the prioritizer prioritize the event.
-    [prioritizer prioritizeEvent:storedEvent];
   });
 }
 
 - (void)removeEvents:(NSSet<GDTStoredEvent *> *)events {
   NSSet<GDTStoredEvent *> *eventsToRemove = [events copy];
-  dispatch_async(_storageQueue, ^{
-    // Check that a prioritizer is available for this target.
-    id<GDTPrioritizer> prioritizer;
+  GDTStoredEvent *anyEvent = [eventsToRemove anyObject];
+  id<GDTPrioritizer> prioritizer =
+      [GDTRegistrar sharedInstance].targetToPrioritizer[anyEvent.target];
+  GDTAssert(prioritizer, @"There must be a prioritizer.");
+  [prioritizer unprioritizeEvents:events];
 
+  dispatch_async(_storageQueue, ^{
     for (GDTStoredEvent *event in eventsToRemove) {
       // Remove from disk, first and foremost.
       NSError *error;
       [[NSFileManager defaultManager] removeItemAtURL:event.eventFileURL error:&error];
       GDTAssert(error == nil, @"There was an error removing an event file: %@", error);
-
-      if (!prioritizer) {
-        prioritizer = [GDTRegistrar sharedInstance].targetToPrioritizer[event.target];
-      } else {
-        GDTAssert(prioritizer == [GDTRegistrar sharedInstance].targetToPrioritizer[event.target],
-                  @"All logs within an upload set should have the same prioritizer.");
-      }
+      GDTAssert([GDTRegistrar sharedInstance].targetToPrioritizer[event.target] == prioritizer,
+                @"All logs within an upload set should have the same prioritizer.");
 
       // Remove from the tracking collections.
       [self.storedEvents removeObject:event];
       [self.targetToEventSet[event.target] removeObject:event];
     }
-    GDTAssert(prioritizer, @"There's no prioritizer registered for the given target.");
-    [prioritizer unprioritizeEvents:events];
   });
 }
 

GoogleDataTransport/GoogleDataTransport/Classes/GDTUploadCoordinator.h

@@ -34,10 +34,9 @@ NS_ASSUME_NONNULL_BEGIN
 /** Forces the backend specified by the target to upload the provided set of events. This should
  * only ever happen when the QoS tier of an event requires it.
  *
- * @param events The set of event hashes to force upload.
  * @param target The target that should force an upload.
  */
-- (void)forceUploadEvents:(NSSet<GDTStoredEvent *> *)events target:(GDTTarget)target;
+- (void)forceUploadForTarget:(GDTTarget)target;
 
 @end
 

GoogleDataTransport/GoogleDataTransport/Classes/GDTUploadCoordinator.m

@@ -51,19 +51,23 @@ - (instancetype)init {
   return self;
 }
 
-- (void)forceUploadEvents:(NSSet<GDTStoredEvent *> *)events target:(GDTTarget)target {
+- (void)forceUploadForTarget:(GDTTarget)target {
   dispatch_async(_coordinationQueue, ^{
     GDTLogWarning(GDTMCWForcedUpload, @"%@", @"A high priority event has caused an upload.");
     NSNumber *targetNumber = @(target);
-    GDTRegistrar *registrar = self->_registrar;
     GDTUploadCoordinatorForceUploadBlock forceUploadBlock = ^{
-      GDTAssert(events.count, @"It doesn't make sense to force upload of 0 events");
-      id<GDTUploader> uploader = registrar.targetToUploader[targetNumber];
-      GDTUploadPackage *package = [[GDTUploadPackage alloc] init];
-      package.events = [events copy];
-      GDTAssert(uploader, @"Target '%@' is missing an implementation", targetNumber);
+      id<GDTPrioritizer> prioritizer = self->_registrar.targetToPrioritizer[targetNumber];
+      id<GDTUploader> uploader = self->_registrar.targetToUploader[targetNumber];
+      GDTAssert(prioritizer && uploader, @"Target '%@' is missing an implementation", targetNumber);
+      GDTUploadConditions conds = [self uploadConditions];
+      conds |= GDTUploadConditionHighPriority;
+      GDTUploadPackage *package = [[prioritizer uploadPackageWithConditions:conds] copy];
+      package.storage = self.storage;
+      NSAssert(package.events && package.events.count,
+               @"A high priority event should produce events to upload.");
+      self->_targetToInFlightEventSet[targetNumber] = package.events;
       [uploader uploadPackage:package onComplete:self.onCompleteBlock];
-      self->_targetToInFlightEventSet[targetNumber] = events;
+      [self->_forcedUploadQueue removeLastObject];
     };
 
     // Enqueue the force upload block if there's an in-flight upload for that target already.
@@ -112,9 +116,10 @@ - (GDTUploaderCompletionBlock)onCompleteBlock {
             GDTUploadCoordinatorForceUploadBlock queuedBlock =
                 [strongSelf->_forcedUploadQueue lastObject];
             if (queuedBlock) {
-              queuedBlock();
+              dispatch_async(strongSelf->_coordinationQueue, ^{
+                queuedBlock();
+              });
             }
-            [strongSelf->_forcedUploadQueue removeLastObject];
           }
         });
       }

GoogleDataTransport/GoogleDataTransport/Classes/Public/GDTPrioritizer.h

@@ -32,6 +32,9 @@ typedef NS_OPTIONS(NSInteger, GDTUploadConditions) {
 
   /** An upload would likely use wifi data. */
   GDTUploadConditionWifiData,
+
+  /** A high priority event has occurred. */
+  GDTUploadConditionHighPriority,
 };
 
 /** This protocol defines the common interface of event prioritization. Prioritizers are

GoogleDataTransport/Tests/Common/Fakes/GDTUploadCoordinatorFake.m

@@ -18,7 +18,7 @@
 
 @implementation GDTUploadCoordinatorFake
 
-- (void)forceUploadEvents:(NSSet<GDTStoredEvent *> *)events target:(GDTTarget)target {
+- (void)forceUploadForTarget:(GDTTarget)target {
   self.forceUploadCalled = YES;
 }
 

GoogleDataTransport/Tests/Integration/GDTIntegrationTest.m

@@ -152,7 +152,7 @@ - (void)testEndToEndEvent {
   [testServer stop];
 }
 
-/** Generates and events a bunch of random events. */
+/** Generates a bunch of random events. */
 - (void)generateEvents {
   for (int i = 0; i < 50; i++) {
     // Choose a random transport, and randomly choose if it's a telemetry event.

GoogleDataTransport/Tests/Unit/GDTUploadCoordinatorTest.m

@@ -79,17 +79,16 @@ - (void)testSharedInstance {
 
 /** Tests that forcing a event upload works. */
 - (void)testForceUploadEvents {
+  GDTTestUploadPackage *uploadPackage = [[GDTTestUploadPackage alloc] init];
+  uploadPackage.events = [GDTEventGenerator generate3StoredEvents];
+  self.prioritizer.uploadPackage = uploadPackage;
   XCTestExpectation *expectation = [self expectationWithDescription:@"uploader will upload"];
   self.uploader.uploadEventsBlock =
       ^(GDTUploadPackage *_Nonnull package, GDTUploaderCompletionBlock _Nonnull completionBlock) {
         [expectation fulfill];
       };
-  NSSet<GDTStoredEvent *> *eventSet = [GDTEventGenerator generate3StoredEvents];
-  XCTAssertNoThrow([[GDTUploadCoordinator sharedInstance] forceUploadEvents:eventSet
-                                                                     target:_target]);
-  dispatch_sync([GDTUploadCoordinator sharedInstance].coordinationQueue, ^{
-    [self waitForExpectations:@[ expectation ] timeout:0.1];
-  });
+  XCTAssertNoThrow([[GDTUploadCoordinator sharedInstance] forceUploadForTarget:_target]);
+  [self waitForExpectations:@[ expectation ] timeout:0.1];
 }
 
 /** Tests forcing an upload while that target currently has a request in flight queues. */
@@ -101,21 +100,20 @@ - (void)testForceUploadEventsEnqueuesIftargetAlreadyHasEventsInFlight {
       ^(GDTUploadPackage *_Nonnull package, GDTUploaderCompletionBlock _Nonnull completionBlock) {
         [expectation fulfill];
       };
-  NSSet<GDTStoredEvent *> *eventSet = [GDTEventGenerator generate3StoredEvents];
+  GDTTestUploadPackage *uploadPackage = [[GDTTestUploadPackage alloc] init];
+  uploadPackage.events = [GDTEventGenerator generate3StoredEvents];
+  self.prioritizer.uploadPackage = uploadPackage;
   dispatch_sync([GDTUploadCoordinator sharedInstance].coordinationQueue, ^{
     [GDTUploadCoordinator sharedInstance].targetToInFlightEventSet[@(self->_target)] =
         [[NSSet alloc] init];
   });
-  XCTAssertNoThrow([[GDTUploadCoordinator sharedInstance] forceUploadEvents:eventSet
-                                                                     target:_target]);
+  XCTAssertNoThrow([[GDTUploadCoordinator sharedInstance] forceUploadForTarget:_target]);
   dispatch_sync([GDTUploadCoordinator sharedInstance].coordinationQueue, ^{
     XCTAssertEqual([GDTUploadCoordinator sharedInstance].forcedUploadQueue.count, 1);
     [GDTUploadCoordinator sharedInstance].onCompleteBlock(
         self.target, [GDTClock clockSnapshotInTheFuture:1000], nil);
   });
-  dispatch_sync([GDTUploadCoordinator sharedInstance].coordinationQueue, ^{
-    [self waitForExpectations:@[ expectation ] timeout:0.1];
-  });
+  [self waitForExpectations:@[ expectation ] timeout:1.0];
 }
 
 /** Tests the timer is running at the desired frequency. */