--- a/mobile/android/services/src/main/java/org/mozilla/gecko/sync/synchronizer/RecordsChannel.java
+++ b/mobile/android/services/src/main/java/org/mozilla/gecko/sync/synchronizer/RecordsChannel.java
@@ -200,16 +200,17 @@ public class RecordsChannel implements
       Logger.error(LOG_TAG, "Got NoStoreDelegateException in RecordsChannel.store(). This should not occur. Aborting.", e);
       delegate.onFlowStoreFailed(this, e, record.guid);
     }
   }
 
   @Override
   public void onFetchFailed(Exception ex) {
     Logger.warn(LOG_TAG, "onFetchFailed. Informing sink, calling for immediate stop.", ex);
+    sink.sourceFailed(ex);
     numFetchFailed.incrementAndGet();
     this.consumer.halt();
     delegate.onFlowFetchFailed(this, ex);
   }
 
   @Override
   public void onFetchedRecord(Record record) {
     numFetched.incrementAndGet();
@@ -222,16 +223,17 @@ public class RecordsChannel implements
     Logger.trace(LOG_TAG, "onFetchCompleted. Stopping consumer once stores are done.");
     Logger.trace(LOG_TAG, "Fetch timestamp is " + fetchEnd);
     this.fetchEnd = fetchEnd;
     this.consumer.queueFilled();
   }
 
   @Override
   public void onBatchCompleted() {
+    this.sink.storeFlush();
   }
 
   @Override
   public void onRecordStoreFailed(Exception ex, String recordGuid) {
     Logger.trace(LOG_TAG, "Failed to store record with guid " + recordGuid);
     numStoreFailed.incrementAndGet();
     this.consumer.stored();
     delegate.onFlowStoreFailed(this, ex, recordGuid);