TT#182200 delay Redis entry deletion until after MySQL transaction

The MySQL INSERT statements to move processed Redis acc records from
Redis to the respective backup/trash MySQL tables are always issued
within a MySQL transaction (med_handler via medmysql_batch_start), but
the deletions from Redis were done immediately. Therefore if mediator
were to abort within a processing loop, the MySQL transaction would be
rolled back after the entries had already been deleted from Redis,
therefore losing the acc entries.

Solve this by using an internal queue for Redis entries to hold the
lists of entries to be deleted until the MySQL transaction is commited.

Change-Id: Ib41d0e2ca722c66f9e078ca31f7e5ca2b9d9fe2d
(cherry picked from commit bbbe2c9a7b)

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)

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);

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);

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);
 }

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 */