Merge branch 'master' of 10.1.1.27:/shared/ctdb/ctdb-master

(This used to be ctdb commit 0fd3bf919b1b8e5aaa98444c306c6770a6a3209f)

2 files changed, 25 insertions, 10 deletions

diff --git a/ctdb/server/ctdb_ltdb_server.c b/ctdb/server/ctdb_ltdb_server.c
index e5437b99a2..e699c2ad80 100644
--- a/ctdb/server/ctdb_ltdb_server.c
+++ b/ctdb/server/ctdb_ltdb_server.c
@@ -83,6 +83,8 @@ static int ctdb_ltdb_store_server(struct ctdb_db_context *ctdb_db,
 	 */
 	if (data.dsize != 0) {
 		keep = true;
+	} else if (header->flags & (CTDB_REC_RO_HAVE_DELEGATIONS|CTDB_REC_RO_HAVE_READONLY)) {
+		keep = true;
 	} else if (ctdb_db->persistent) {
 		keep = true;
 	} else if (header->flags & CTDB_REC_FLAG_AUTOMATIC) {
diff --git a/ctdb/server/ctdb_persistent.c b/ctdb/server/ctdb_persistent.c
index dd8d4793fb..0f4f4da52d 100644
--- a/ctdb/server/ctdb_persistent.c
+++ b/ctdb/server/ctdb_persistent.c
@@ -456,12 +456,12 @@ static int ctdb_persistent_store(struct ctdb_persistent_write_state *state)
 	}
 
 	for (i=0;i<m->count;i++) {
-		struct ctdb_ltdb_header oldheader;
-		struct ctdb_ltdb_header header;
+		struct ctdb_ltdb_header *oldheader;
+		struct ctdb_ltdb_header *header;
 		TDB_DATA key, data, olddata;
 		TALLOC_CTX *tmp_ctx = talloc_new(state);
 
-		rec = ctdb_marshall_loop_next(m, rec, NULL, &header, &key, &data);
+		rec = ctdb_marshall_loop_next(m, rec, NULL, NULL, &key, &data);
 		
 		if (rec == NULL) {
 			DEBUG(DEBUG_ERR,("Failed to get next record %d for db_id 0x%08x in ctdb_persistent_store\n",
@@ -469,29 +469,42 @@ static int ctdb_persistent_store(struct ctdb_persistent_write_state *state)
 			talloc_free(tmp_ctx);
 			goto failed;			
 		}
+		header = (struct ctdb_ltdb_header *)&data.dptr[0];
 
 		/* fetch the old header and ensure the rsn is less than the new rsn */
-		ret = ctdb_ltdb_fetch(state->ctdb_db, key, &oldheader, tmp_ctx, &olddata);
-		if (ret != 0) {
+		olddata = tdb_fetch(state->ctdb_db->ltdb->tdb, key);
+		if (olddata.dptr == NULL) {
 			DEBUG(DEBUG_ERR,("Failed to fetch old record for db_id 0x%08x in ctdb_persistent_store\n",
 					 state->ctdb_db->db_id));
 			talloc_free(tmp_ctx);
 			goto failed;
 		}
+		if (olddata.dsize < sizeof(struct ctdb_ltdb_header)) {
+			DEBUG(DEBUG_ERR,("Not enough header for record for db_id 0x%08x in ctdb_persistent_store\n",
+					state->ctdb_db->db_id));
+			talloc_free(tmp_ctx);
+			free(olddata.dptr);
+			goto failed;
+		}
+		oldheader = (struct ctdb_ltdb_header *)&olddata.dptr[0];
 
-		if (oldheader.rsn >= header.rsn &&
-		    (olddata.dsize != data.dsize || 
-		     memcmp(olddata.dptr, data.dptr, data.dsize) != 0)) {
+		if (oldheader->rsn >= header->rsn &&
+		   (olddata.dsize != data.dsize || 
+		   memcmp(&olddata.dptr[sizeof(struct ctdb_ltdb_header)],
+		     &data.dptr[sizeof(struct ctdb_ltdb_header)],
+		     data.dsize - sizeof(struct ctdb_ltdb_header)) != 0)) {
 			DEBUG(DEBUG_CRIT,("existing header for db_id 0x%08x has larger RSN %llu than new RSN %llu in ctdb_persistent_store\n",
 					  state->ctdb_db->db_id, 
-					  (unsigned long long)oldheader.rsn, (unsigned long long)header.rsn));
+					  (unsigned long long)oldheader->rsn, (unsigned long long)header->rsn));
 			talloc_free(tmp_ctx);
+			free(olddata.dptr);
 			goto failed;
 		}
 
 		talloc_free(tmp_ctx);
+		free(olddata.dptr);
 
-		ret = ctdb_ltdb_store(state->ctdb_db, key, &header, data);
+		ret = tdb_store(state->ctdb_db->ltdb->tdb, key, data, TDB_REPLACE);
 		if (ret != 0) {
 			DEBUG(DEBUG_CRIT,("Failed to store record for db_id 0x%08x in ctdb_persistent_store\n", 
 					  state->ctdb_db->db_id));