From bf1d429227dc4f5818263cc39401d0a22663cdba Mon Sep 17 00:00:00 2001
From: Ronnie Sahlberg <ronniesahlberg@gmail.com>
Date: Mon, 24 Oct 2011 13:14:26 +1100
Subject: [PATCH] ReadOnly: Dont update the record header from the calling
 client. While it is convenient since it avoids having to create a child
 process from the main dameon for writing the updated record it makes the
 cleitn more complex.

Remove the code in the example client code that writes the record to the local tdb.
Add code to the local ctdbd processing of replies to check if this reply contain a ro delegation and if so, spawn a child process to lock the tdb and then write the data.
---
 client/ctdb_client.c | 26 ---------------------
 server/ctdb_call.c   | 54 ++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 54 insertions(+), 26 deletions(-)

diff --git a/client/ctdb_client.c b/client/ctdb_client.c
index f2359caf..f0172e42 100644
--- a/client/ctdb_client.c
+++ b/client/ctdb_client.c
@@ -843,32 +843,6 @@ again:
 			goto again;
 		}
 
-		if (h->header.rsn >= roheader->rsn) {
-			DEBUG(DEBUG_ERR,("READONLY RECORD: Too small RSN, migrate and try again\n"));
-			ctdb_ltdb_unlock(ctdb_db, key);
-
-			ret = ctdb_client_force_migration(ctdb_db, key);
-			if (ret != 0) {
-				DEBUG(DEBUG_DEBUG,("ctdb_fetch_readonly_lock: force_migration failed\n"));
-				talloc_free(h);
-				return NULL;
-			}
-
-			goto again;
-		}
-
-		if (ctdb_ltdb_store(ctdb_db, key, roheader, rodata) != 0) {
-			ctdb_ltdb_unlock(ctdb_db, key);
-
-			ret = ctdb_client_force_migration(ctdb_db, key);
-			if (ret != 0) {
-				DEBUG(DEBUG_DEBUG,("ctdb_fetch_readonly_lock: force_migration failed\n"));
-				talloc_free(h);
-				return NULL;
-			}
-
-			goto again;
-		}
 		return h;
 	}
 
diff --git a/server/ctdb_call.c b/server/ctdb_call.c
index 9e864222..9fc8b339 100644
--- a/server/ctdb_call.c
+++ b/server/ctdb_call.c
@@ -700,6 +700,60 @@ void ctdb_reply_call(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
 		return;
 	}
 
+
+	/* read only delegation processing */
+	/* If we got a FETCH_WITH_HEADER we should check if this is a ro
+	 * delegation since we may need to update the record header
+	 */
+	if (state->c->callid == CTDB_FETCH_WITH_HEADER_FUNC) {
+		struct ctdb_db_context *ctdb_db = state->ctdb_db;
+		struct ctdb_ltdb_header *header = (struct ctdb_ltdb_header *)&c->data[0];
+		struct ctdb_ltdb_header oldheader;
+		TDB_DATA key, data, olddata;
+		int ret;
+
+		if (!(header->flags & CTDB_REC_RO_HAVE_READONLY)) {
+			goto finished_ro;
+			return;
+		}
+
+		key.dsize = state->c->keylen;
+		key.dptr  = state->c->data;
+		ret = ctdb_ltdb_lock_requeue(ctdb_db, key, hdr,
+				     ctdb_call_input_pkt, ctdb, False);
+		if (ret == -2) {
+			return;
+		}
+		if (ret != 0) {
+			DEBUG(DEBUG_ERR,(__location__ " Failed to get lock in ctdb_reply_call\n"));
+			return;
+		}
+
+		ret = ctdb_ltdb_fetch(ctdb_db, key, &oldheader, state, &olddata);
+		if (ret != 0) {
+			DEBUG(DEBUG_ERR, ("Failed to fetch old record in ctdb_reply_call\n"));
+			ctdb_ltdb_unlock(ctdb_db, key);
+			goto finished_ro;
+		}			
+
+		if (header->rsn <= oldheader.rsn) {
+			ctdb_ltdb_unlock(ctdb_db, key);
+			goto finished_ro;
+		}
+
+		data.dptr  = &c->data[sizeof(struct ctdb_ltdb_header)];
+		data.dsize = sizeof(struct ctdb_ltdb_header);
+		ret = ctdb_ltdb_store(ctdb_db, key, header, data);
+		if (ret != 0) {
+			DEBUG(DEBUG_ERR, ("Failed to store new record in ctdb_reply_call\n"));
+			ctdb_ltdb_unlock(ctdb_db, key);
+			goto finished_ro;
+		}			
+
+		ctdb_ltdb_unlock(ctdb_db, key);
+	}
+finished_ro:
+
 	state->call->reply_data.dptr = c->data;
 	state->call->reply_data.dsize = c->datalen;
 	state->call->status = c->status;
-- 
2.34.1