Add a new header flag for "migrated with data" and set this to 1

when we migrate a non-empty record onto the node
or a non-empty record off the node

When we migrate a record back to the lmaster and yield the dmaster role,
inspect this flag if if it is still not set, we can delete the record from
the local database as soon as we have migrated it back to the lmaster.

diff --git a/server/ctdb_call.c b/server/ctdb_call.c
index ed741ed0a9d5f4cd7a6b9bf26836b4ee8760265d..a675e7f2eb058abaf35cbd26987bfc6b3fc2c50c 100644 (file)
--- a/server/ctdb_call.c
+++ b/server/ctdb_call.c
@@ -202,9 +202,10 @@ static void ctdb_call_send_dmaster(struct ctdb_db_context *ctdb_db,
        }
 
        if (data->dsize != 0) {
-               header->flags &= CTDB_REC_FLAG_MIGRATED_WITH_DATA;
+               header->flags |= CTDB_REC_FLAG_MIGRATED_WITH_DATA;
        }
 
+
        if (lmaster == ctdb->pnn) {
                ctdb_send_dmaster_reply(ctdb_db, header, *key, *data, 
                                        c->hdr.srcnode, c->hdr.reqid);
@@ -226,10 +227,19 @@ static void ctdb_call_send_dmaster(struct ctdb_db_context *ctdb_db,
        memcpy(&r->data[key->dsize], data->dptr, data->dsize);
 
        header->dmaster = c->hdr.srcnode;
-       if (ctdb_ltdb_store(ctdb_db, *key, header, *data) != 0) {
-               ctdb_fatal(ctdb, "Failed to store record in ctdb_call_send_dmaster");
+
+       if (data->dsize == 0
+       && lmaster != ctdb->pnn
+       && (header->flags & CTDB_REC_FLAG_MIGRATED_WITH_DATA) == 0) {
+               if (ctdb_ltdb_delete(ctdb_db, *key) != 0) {
+                       ctdb_fatal(ctdb, "Failed to delete empty record when migrating it off the node");
+               }
+       } else {
+               if (ctdb_ltdb_store(ctdb_db, *key, header, *data) != 0) {
+                       ctdb_fatal(ctdb, "Failed to store record in ctdb_call_send_dmaster");
+               }
        }
-       
+
        ctdb_queue_packet(ctdb, &r->hdr);
 
        talloc_free(r);
@@ -257,6 +267,10 @@ static void ctdb_become_dmaster(struct ctdb_db_context *ctdb_db,
        header.rsn = rsn + 1;
        header.dmaster = ctdb->pnn;
 
+       if (data.dsize != 0) {
+               header.flags |= CTDB_REC_FLAG_MIGRATED_WITH_DATA;
+       }
+
        if (ctdb_ltdb_store(ctdb_db, key, &header, data) != 0) {
                ctdb_fatal(ctdb, "ctdb_reply_dmaster store failed\n");
 
@@ -355,6 +369,10 @@ void ctdb_request_dmaster(struct ctdb_context *ctdb, struct ctdb_req_header *hdr
                return;
        }
 
+       if (data.dsize != 0) {
+               header.flags |= CTDB_REC_FLAG_MIGRATED_WITH_DATA;
+       }
+
        if (ctdb_lmaster(ctdb, &key) != ctdb->pnn) {
                DEBUG(DEBUG_ALERT,("pnn %u dmaster request to non-lmaster lmaster=%u gen=%u curgen=%u\n",
                         ctdb->pnn, ctdb_lmaster(ctdb, &key),