diff --git a/cdr.c b/cdr.c
index d0cfcc7..947b4ac 100644
--- a/cdr.c
+++ b/cdr.c
@@ -62,9 +62,11 @@ static void free_cdrs(cdr_entry_t **cdrs, uint64_t cdr_count) {
 }
 
 
-int cdr_process_records(med_entry_t *records, uint64_t count, uint64_t *ext_count,
+int cdr_process_records(med_entry_t **records_p, uint64_t *count_p, uint64_t *ext_count,
         struct medmysql_batches *batches, int do_intermediate)
 {
+    med_entry_t *records = *records_p;
+    uint64_t count = *count_p;
     int ret = 0;
     uint8_t trash = 0;
     uint64_t i;
@@ -162,7 +164,7 @@ int cdr_process_records(med_entry_t *records, uint64_t count, uint64_t *ext_coun
                         {
                             if (has_redis)
                             {
-                                if(medredis_backup_entries(records, count) != 0)
+                                if(medredis_backup_entries(records_p, count_p) != 0)
                                     goto error;
                             }
                             if (has_mysql)
@@ -210,7 +212,7 @@ int cdr_process_records(med_entry_t *records, uint64_t count, uint64_t *ext_coun
     {
         if (has_redis)
         {
-            if(medredis_trash_entries(records, count) != 0)
+            if(medredis_trash_entries(records_p, count_p) != 0)
                 goto error;
         }
         if (has_mysql)
diff --git a/cdr.h b/cdr.h
index ade2397..db65a7c 100644
--- a/cdr.h
+++ b/cdr.h
@@ -59,7 +59,7 @@ typedef struct {
 #undef F
 #undef FA
 
-int cdr_process_records(med_entry_t *records, uint64_t count, uint64_t *cdr_count, struct medmysql_batches *,
+int cdr_process_records(med_entry_t **records, uint64_t *count, uint64_t *cdr_count, struct medmysql_batches *,
         int do_intermediate);
 void cdr_fix_accids(med_entry_t *records, uint64_t count);
 int cdr_fill_record(cdr_entry_t *cdr);
diff --git a/mediator.c b/mediator.c
index 3ff146c..aee975f 100644
--- a/mediator.c
+++ b/mediator.c
@@ -370,7 +370,7 @@ int main(int argc, char **argv)
                 continue;
             }
 
-            if(cdr_process_records(mysql_records, mysql_rec_count, &cdr_count, batches, do_intermediate) != 0)
+            if(cdr_process_records(&mysql_records, &mysql_rec_count, &cdr_count, batches, do_intermediate) != 0)
                 goto out;
 
             if(mysql_rec_count > 0)
@@ -436,7 +436,7 @@ int main(int argc, char **argv)
             L_DEBUG("process cdr with cid '%s' and %"PRIu64" records\n", redis_callids[i].value, redis_rec_count);
 
             if (redis_rec_count) {
-                if(cdr_process_records(redis_records, redis_rec_count, &cdr_count, batches, do_intermediate) != 0) {
+                if(cdr_process_records(&redis_records, &redis_rec_count, &cdr_count, batches, do_intermediate) != 0) {
                     free(redis_records);
                     goto out;
                 }
@@ -460,6 +460,8 @@ int main(int argc, char **argv)
 
         if (medmysql_batch_end(batches))
             break;
+        if (medredis_batch_end())
+            break;
 
         gettimeofday(&loop_tv_stop, NULL);
         loop_runtime = mediator_calc_runtime(&loop_tv_start, &loop_tv_stop);
diff --git a/medredis.c b/medredis.c
index a47382e..d2af9a4 100644
--- a/medredis.c
+++ b/medredis.c
@@ -44,15 +44,21 @@ typedef struct {
     uint64_t index;
 } medredis_cidlist_t;
 
+typedef struct {
+    med_entry_t *records;
+    uint64_t count;
+} medredis_records_list_t;
+
 typedef struct {
     const char *name;
+    GQueue record_lists;
 } medredis_table_t;
 
 static medredis_con_t *con = NULL;
 static char medredis_srem_key_lua[41]; // sha-1 hex string
 
-static medredis_table_t medredis_table_trash = { .name = "trash" };
-static medredis_table_t medredis_table_backup = { .name = "backup" };
+static medredis_table_t medredis_table_trash = { .name = "trash", .record_lists = G_QUEUE_INIT };
+static medredis_table_t medredis_table_backup = { .name = "backup", .record_lists = G_QUEUE_INIT };
 
 /**********************************************************************/
 static void medredis_free_reply(redisReply **reply) {
@@ -829,16 +835,34 @@ err:
 }
 
 /**********************************************************************/
-static int medredis_cleanup_entries(med_entry_t *records, uint64_t count, medredis_table_t *table) {
-    char buffer[512];
-
-    if (medmysql_insert_records(records, count, table->name) != 0) {
+static int medredis_cleanup_entries(med_entry_t **records, uint64_t *count, medredis_table_t *table) {
+    if (medmysql_insert_records(*records, *count, table->name) != 0) {
         L_CRITICAL("Failed to cleanup redis records\n");
         goto err;
     }
 
-    for (uint64_t i = 0; i < count; ++i) {
-        med_entry_t *e = &(records[i]);
+    // take ownership of the contents of the `records` array and swap out contents with
+    // an empty array
+    medredis_records_list_t *list = g_slice_alloc(sizeof(*list));
+    list->records = *records;
+    list->count = *count;
+    *count = 0;
+    *records = NULL;
+
+    g_queue_push_tail(&table->record_lists, list);
+
+    return 0;
+
+err:
+    return -1;
+}
+
+/**********************************************************************/
+static int medredis_batch_end_records(medredis_records_list_t *list) {
+    char buffer[512];
+
+    for (uint64_t i = 0; i < list->count; ++i) {
+        med_entry_t *e = &(list->records[i]);
         if (!e->redis)
             continue;
 
@@ -880,11 +904,33 @@ err:
 }
 
 /**********************************************************************/
-int medredis_trash_entries(med_entry_t *records, uint64_t count) {
+static int medredis_batch_end_table(medredis_table_t *table) {
+    while (table->record_lists.length) {
+        medredis_records_list_t *list = g_queue_pop_head(&table->record_lists);
+        int ret = medredis_batch_end_records(list);
+        free(list->records);
+        g_slice_free1(sizeof(*list), list);
+        if (ret)
+            return -1;
+    }
+    return 0;
+}
+
+/**********************************************************************/
+int medredis_batch_end(void) {
+    if (medredis_batch_end_table(&medredis_table_trash))
+        return -1;
+    if (medredis_batch_end_table(&medredis_table_backup))
+        return -1;
+    return 0;
+}
+
+/**********************************************************************/
+int medredis_trash_entries(med_entry_t **records, uint64_t *count) {
     return medredis_cleanup_entries(records, count, &medredis_table_trash);
 }
 
 /**********************************************************************/
-int medredis_backup_entries(med_entry_t *records, uint64_t count) {
+int medredis_backup_entries(med_entry_t **records, uint64_t *count) {
     return medredis_cleanup_entries(records, count, &medredis_table_backup);
 }
diff --git a/medredis.h b/medredis.h
index 3d39e44..9d2a485 100644
--- a/medredis.h
+++ b/medredis.h
@@ -10,7 +10,8 @@ int medredis_init(void);
 void medredis_cleanup(void);
 med_callid_t *medredis_fetch_callids(uint64_t *count);
 int medredis_fetch_records(med_callid_t *callid, med_entry_t **entries, uint64_t *count);
-int medredis_trash_entries(med_entry_t *records, uint64_t count);
-int medredis_backup_entries(med_entry_t *records, uint64_t count);
+int medredis_trash_entries(med_entry_t **records, uint64_t *count);
+int medredis_backup_entries(med_entry_t **records, uint64_t *count);
+int medredis_batch_end(void);
 
 #endif /* _MED_REDIS_H */