- accept an optional set of tdb_flags from clients on open a database,

thus allowing the client to pass through the TDB_NOSYNC flag

- ensure that tdb_store() operations on persistent databases that don't
  have TDB_NOSYNC set happen inside a transaction wrapper, thus making
  them crash safe

(This used to be ctdb commit 49330f97c78ca0669615297ac3d8498651831214)

ctdb/common/ctdb_ltdb.c

@@ -150,7 +150,25 @@ int ctdb_ltdb_store(struct ctdb_db_context *ctdb_db, TDB_DATA key,
 	memcpy(rec.dptr, header, sizeof(*header));
 	memcpy(rec.dptr + sizeof(*header), data.dptr, data.dsize);
 
-	ret = tdb_store(ctdb_db->ltdb->tdb, key, rec, TDB_REPLACE);
+	/* if this is a persistent database without NOSYNC then we
+	   will do this via a transaction */
+	if (ctdb_db->persistent && !(ctdb_db->client_tdb_flags & TDB_NOSYNC)) {
+		ret = tdb_transaction_start(ctdb_db->ltdb->tdb);
+		if (ret != 0) {
+			DEBUG(DEBUG_CRIT, ("Failed to start local transaction\n"));
+			goto failed;
+		}
+		ret = tdb_store(ctdb_db->ltdb->tdb, key, rec, TDB_REPLACE);
+		if (ret != 0) {
+			tdb_transaction_cancel(ctdb_db->ltdb->tdb);
+			goto failed;
+		}
+		ret = tdb_transaction_commit(ctdb_db->ltdb->tdb);
+	} else {
+		ret = tdb_store(ctdb_db->ltdb->tdb, key, rec, TDB_REPLACE);
+	}
+
+failed:
 	talloc_free(rec.dptr);
 
 	return ret;

ctdb/include/ctdb_private.h

@@ -400,6 +400,7 @@ struct ctdb_db_context {
 	struct ctdb_registered_call *calls; /* list of registered calls */
 	uint32_t seqnum;
 	struct timed_event *te;
+	uint32_t client_tdb_flags;
 };
 
 
@@ -910,7 +911,7 @@ int ctdb_daemon_send_control(struct ctdb_context *ctdb, uint32_t destnode,
 			     void *private_data);
 
 int32_t ctdb_control_db_attach(struct ctdb_context *ctdb, TDB_DATA indata, 
-			       TDB_DATA *outdata, bool persistent);
+			       TDB_DATA *outdata, uint64_t tdb_flags, bool persistent);
 
 int ctdb_daemon_set_call(struct ctdb_context *ctdb, uint32_t db_id,
 			 ctdb_fn_t fn, int id);

ctdb/server/ctdb_control.c

@@ -206,10 +206,10 @@ static int32_t ctdb_control_dispatch(struct ctdb_context *ctdb,
 	}
 
 	case CTDB_CONTROL_DB_ATTACH:
-		return ctdb_control_db_attach(ctdb, indata, outdata, false);
+		return ctdb_control_db_attach(ctdb, indata, outdata, srvid, false);
 
 	case CTDB_CONTROL_DB_ATTACH_PERSISTENT:
-		return ctdb_control_db_attach(ctdb, indata, outdata, true);
+		return ctdb_control_db_attach(ctdb, indata, outdata, srvid, true);
 
 	case CTDB_CONTROL_SET_CALL: {
 		struct ctdb_control_set_call *sc = 

ctdb/server/ctdb_ltdb_server.c

@@ -296,12 +296,19 @@ static int ctdb_local_attach(struct ctdb_context *ctdb, const char *db_name, boo
   a client has asked to attach a new database
  */
 int32_t ctdb_control_db_attach(struct ctdb_context *ctdb, TDB_DATA indata,
-			       TDB_DATA *outdata, bool persistent)
+			       TDB_DATA *outdata, uint64_t tdb_flags, 
+			       bool persistent)
 {
 	const char *db_name = (const char *)indata.dptr;
 	struct ctdb_db_context *db;
 	struct ctdb_node *node = ctdb->nodes[ctdb->pnn];
 
+	/* the client can optionally pass additional tdb flags, but we
+	   only allow a subset of those on the database in ctdb. Note
+	   that tdb_flags is passed in via the (otherwise unused)
+	   srvid to the attach control */
+	tdb_flags &= TDB_NOSYNC;
+
 	/* If the node is inactive it is not part of the cluster
 	   and we should not allow clients to attach to any
 	   databases
@@ -317,6 +324,7 @@ int32_t ctdb_control_db_attach(struct ctdb_context *ctdb, TDB_DATA indata,
 	if (db) {
 		outdata->dptr  = (uint8_t *)&db->db_id;
 		outdata->dsize = sizeof(db->db_id);
+		db->client_tdb_flags |= tdb_flags;
 		return 0;
 	}
 
@@ -330,6 +338,9 @@ int32_t ctdb_control_db_attach(struct ctdb_context *ctdb, TDB_DATA indata,
 		return -1;
 	}
 
+	/* remember the flags the client has specified */
+	db->client_tdb_flags = tdb_flags;
+
 	outdata->dptr  = (uint8_t *)&db->db_id;
 	outdata->dsize = sizeof(db->db_id);