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 f3d93a8d972001abbd272db114972c3125304c7c..7bc4c351a8acd6f320376097550c79796e58ce20 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),