memtx: make user changes transactional

There are a few issues concerning TX manager and DDL operations.

The thing is metadata stored for any object is not currently designed to
be versioned. It means that several transactions can see changes
provided by each other BEFORE they are committed. Which is considered
to be a disaster and violates all transaction processing principles
(speaking of serializable isolation level).

Regarding users in particular, there's quite simple test case leading to
the crash. Imagine two transaction creating a user with the same ID at the
same moment:

tx1:begin()
tx2:begin()

tx1("box.schema.user.create('internal1')")
tx2("box.schema.user.create('internal2')")

First operation creates and adds user with ID_1 to user cache; second
operation finds user by the same ID_1 and attempts to update it (to be
more precise - its name). We get the same ID since ID for user is
generated based on the last value appearing in the space; before
transaction is committed the last value for both transactions is the
same.

In on_replace_dd_user() trigger we assume that the operation is
update/replace if we find user in the cache by its id (in contrast to
other on_replace DDL triggers where we look at presence of old_tuple).
Unfortunately on_rollback trigger supposes that old_tuple is not null.
Meanwhile it is: until first TX commits its changes, old_tuple does not
exist.

In scope of this patch we are going to fix several problems at once.

Firstly, let's add to the struct user id of the transaction that
attempts at modifying it. It is assigned after change in
on_replace_dd_user() and reset to zero in on_commit/on_rollback triggers.
If any other transaction (with different id) wants to acquire the same
user - we raise an error saying that user is already busy. To deal with
the situation when TX deletes user, we also add is_deleted flag which
indicates that user has been deleted but transaction is still not
committed.

Secondly, let's rework on_rollback and introduce on_commit triggers of
on_replace_dd_user(). Now they accept struct user or user_def and reset
tx id of user.

Obviously, that patch doesn't fix all the problems. In fact, users have
quite sophisticated implementation. They are stored in two places: as an
static array and in cache. Furthermore, we maintain dependency graph for
roles and RB-Tree of privileges for each user. That makes tracking of
privilege updates complicated. So that we don't touch this part in scope
of current patch.

Part of #6138
Closes #5998

src/box/alter.cc

@@ -3090,30 +3090,75 @@ user_def_new_from_tuple(struct tuple *tuple)
 }
 
 static int
-user_cache_remove_user(struct trigger *trigger, void * /* event */)
+on_create_user_commit(struct trigger * trigger, void * /* event */)
 {
-	struct tuple *tuple = (struct tuple *)trigger->data;
-	uint32_t uid;
-	if (tuple_field_u32(tuple, BOX_USER_FIELD_ID, &uid) != 0)
-		return -1;
-	user_cache_delete(uid);
+	struct user *user = (struct user *)trigger->data;
+	user->tx_id = 0;
 	return 0;
 }
 
 static int
-user_cache_alter_user(struct trigger *trigger, void * /* event */)
+on_create_user_rollback(struct trigger * trigger, void * /* event */)
 {
-	struct tuple *tuple = (struct tuple *)trigger->data;
-	struct user_def *user = user_def_new_from_tuple(tuple);
-	if (user == NULL)
-		return -1;
-	if (user_cache_replace(user) == NULL) {
-		free(user);
-		return -1;
-	}
+	struct user *user = (struct user *)trigger->data;
+	user_cache_delete(user->def->uid);
+	return 0;
+}
+
+static int
+on_delete_user_commit(struct trigger * trigger, void * /* event */)
+{
+	struct user *user = (struct user *)trigger->data;
+	if (user->is_deleted)
+		user_cache_delete(user->def->uid);
+	return 0;
+}
+
+static int
+on_delete_user_rollback(struct trigger * trigger, void * /* event */)
+{
+	struct user *user = (struct user *)trigger->data;
+	user->is_deleted = false;
+	user->tx_id = 0;
+	return 0;
+}
+
+static int
+on_alter_user_commit(struct trigger * trigger, void * /* event */)
+{
+	struct user_def *old_def = (struct user_def *)trigger->data;
+	struct user *new_user = user_by_id(old_def->uid);
+	assert(new_user != NULL);
+	new_user->tx_id = 0;
+	new_user->is_deleted = false;
+	free(old_def);
+	return 0;
+}
+
+static int
+on_alter_user_rollback(struct trigger * trigger, void * /* event */)
+{
+	struct user_def *old_def = (struct user_def *)trigger->data;
+	struct user *new_user = user_by_id(old_def->uid);
+	assert(new_user != NULL);
+	struct user *old_user = user_cache_replace(old_def);
+	assert(old_user != NULL);
+	old_user->tx_id = 0;
+	assert(!old_user->is_deleted);
 	return 0;
 }
 
+static struct user_def *
+user_def_dup(struct user_def *user)
+{
+	size_t def_size = user_def_sizeof(strlen(user->name));
+	struct user_def *dup = (struct user_def *) malloc(def_size);
+	if (dup == NULL)
+		return NULL;
+	memcpy(dup, user, def_size);
+	return dup;
+}
+
 /**
  * A trigger invoked on replace in the user table.
  */
@@ -3130,22 +3175,32 @@ on_replace_dd_user(struct trigger * /* trigger */, void *event)
 			    BOX_USER_FIELD_ID, &uid) != 0)
 		return -1;
 	struct user *old_user = user_by_id(uid);
+	if (old_user != NULL && !user_check_tx_ownership(old_user)) {
+		diag_set(ClientError, ER_USER_BUSY);
+		return -1;
+	}
 	if (new_tuple != NULL && old_user == NULL) { /* INSERT */
 		struct user_def *user = user_def_new_from_tuple(new_tuple);
 		if (user == NULL)
 			return -1;
 		if (access_check_ddl(user->name, user->uid, user->owner, user->type,
 				 PRIV_C) != 0)
 			return -1;
-		if (user_cache_replace(user) == NULL) {
+		struct user *new_user = user_cache_replace(user);
+		if (new_user == NULL) {
 			free(user);
 			return -1;
 		}
+		new_user->tx_id = in_txn()->id;
 		struct trigger *on_rollback =
-			txn_alter_trigger_new(user_cache_remove_user, new_tuple);
-		if (on_rollback == NULL)
+			txn_alter_trigger_new(on_create_user_rollback,
+					      new_user);
+		struct trigger *on_commit =
+			txn_alter_trigger_new(on_create_user_commit, new_user);
+		if (on_rollback == NULL || on_commit == NULL)
 			return -1;
 		txn_stmt_on_rollback(stmt, on_rollback);
+		txn_stmt_on_commit(stmt, on_commit);
 	} else if (new_tuple == NULL) { /* DELETE */
 		if (access_check_ddl(old_user->def->name, old_user->def->uid,
 				 old_user->def->owner, old_user->def->type,
@@ -3171,12 +3226,27 @@ on_replace_dd_user(struct trigger * /* trigger */, void *event)
 				  old_user->def->name, "the user has objects");
 			return -1;
 		}
-		user_cache_delete(uid);
+		struct user *to_delete = user_by_id(uid);
+		/*
+		 * We can't delete user twice: if another transaction attempts
+		 * at deleting this user tx id check will fire first;
+		 * if we delete user twice in the same transaction, then MVCC
+		 * won't find corresponding tuple and on_replace trigger won't
+		 * fire.
+		 */
+		assert(!to_delete->is_deleted);
+		to_delete->is_deleted = true;
+		to_delete->tx_id = in_txn()->id;
 		struct trigger *on_rollback =
-			txn_alter_trigger_new(user_cache_alter_user, old_tuple);
-		if (on_rollback == NULL)
+			txn_alter_trigger_new(on_delete_user_rollback,
+					      to_delete);
+		struct trigger *on_commit =
+			txn_alter_trigger_new(on_delete_user_commit,
+					      to_delete);
+		if (on_rollback == NULL || on_commit == NULL)
 			return -1;
 		txn_stmt_on_rollback(stmt, on_rollback);
+		txn_stmt_on_commit(stmt, on_commit);
 	} else { /* UPDATE, REPLACE */
 		assert(old_user != NULL && new_tuple != NULL);
 		/*
@@ -3192,15 +3262,26 @@ on_replace_dd_user(struct trigger * /* trigger */, void *event)
 			free(user);
 			return -1;
 		}
-		if (user_cache_replace(user) == NULL) {
+		struct user_def *old_def_dup = user_def_dup(old_user->def);
+		struct user *replaced = user_cache_replace(user);
+		if (replaced == NULL) {
 			free(user);
 			return -1;
 		}
+		replaced->tx_id = in_txn()->id;
+		/* In fact sequence of DELETE + INSERT request inside one
+		 * transaction leads to DELETE + REPLACE since DELETE doesn't
+		 * erase entry from the cache until on_commit trigger is fired.
+		 */
+		replaced->is_deleted = false;
 		struct trigger *on_rollback =
-			txn_alter_trigger_new(user_cache_alter_user, old_tuple);
-		if (on_rollback == NULL)
+			txn_alter_trigger_new(on_alter_user_rollback, old_def_dup);
+		struct trigger *on_commit =
+			txn_alter_trigger_new(on_alter_user_commit, old_def_dup);
+		if (on_rollback == NULL || on_commit == NULL)
 			return -1;
 		txn_stmt_on_rollback(stmt, on_rollback);
+		txn_stmt_on_commit(stmt, on_commit);
 	}
 	return 0;
 }
@@ -4037,7 +4118,7 @@ grant_or_revoke(struct priv_def *priv)
 	 * and the role is specified.
 	 */
 	if (priv->object_type == SC_ROLE && !(priv->access & ~PRIV_X)) {
-		struct user *role = user_by_id(priv->object_id);
+		struct user *role = user_find(priv->object_id);
 		if (role == NULL || role->def->type != SC_ROLE)
 			return 0;
 		if (priv->access) {

src/box/errcode.h

@@ -277,6 +277,7 @@ struct errcode_record {
 	/*222 */_(ER_QUORUM_WAIT,		"Couldn't wait for quorum %d: %s") \
 	/*223 */_(ER_INTERFERING_PROMOTE,	"Instance with replica id %u was promoted first") \
 	/*224 */_(ER_RAFT_DISABLED,		"Elections were turned off while running box.ctl.promote()")\
+	/*225 */_(ER_USER_BUSY,			"Can't modify user: another transaction has already acquired it") \
 
 /*
  * !IMPORTANT! Please follow instructions at start of the file

src/box/user.cc

@@ -39,6 +39,7 @@
 #include "scoped_guard.h"
 #include "sequence.h"
 #include "tt_static.h"
+#include "txn.h"
 
 struct universe universe;
 static struct user users[BOX_USER_MAX];
@@ -165,6 +166,8 @@ user_create(struct user *user, uint8_t auth_token)
 	privset_new(&user->privs);
 	region_create(&user->pool, &cord()->slabc);
 	rlist_create(&user->credentials_list);
+	user->is_deleted = false;
+	user->tx_id = 0;
 }
 
 static void
@@ -502,6 +505,16 @@ user_cache_delete(uint32_t uid)
 	}
 }
 
+bool
+user_check_tx_ownership(struct user *user)
+{
+	if (user->tx_id == 0)
+		return true;
+	if (in_txn() == NULL)
+		return true;
+	return user->tx_id == in_txn()->id;
+}
+
 /** Find user by id. */
 struct user *
 user_by_id(uint32_t uid)
@@ -516,8 +529,18 @@ struct user *
 user_find(uint32_t uid)
 {
 	struct user *user = user_by_id(uid);
-	if (user == NULL)
+	if (user == NULL) {
 		diag_set(ClientError, ER_NO_SUCH_USER, int2str(uid));
+		return NULL;
+	}
+	if (!user_check_tx_ownership(user)) {
+		diag_set(ClientError, ER_USER_BUSY);
+		return NULL;
+	}
+	if (user->is_deleted) {
+		diag_set(ClientError, ER_NO_SUCH_USER, int2str(uid));
+		return NULL;
+	}
 	return user;
 }
 
@@ -536,9 +559,11 @@ user_find_by_name(const char *name, uint32_t len)
 	if (schema_find_id(BOX_USER_ID, 2, name, len, &uid) != 0)
 		return NULL;
 	if (uid != BOX_ID_NIL) {
-		struct user *user = user_by_id(uid);
+		struct user *user = user_find(uid);
 		if (user != NULL && user->def->type == SC_USER)
 			return user;
+		if (user == NULL)
+			return NULL;
 	}
 	diag_set(ClientError, ER_NO_SUCH_USER,
 		 tt_cstr(name, MIN((uint32_t) BOX_INVALID_NAME_MAX, len)));

src/box/user.h

@@ -73,6 +73,16 @@ rb_proto(, privset_, privset_t, struct priv_def);
 struct user
 {
 	struct user_def *def;
+	/**
+	 * Id of the transaction that processes this user. Is set only during
+	 * DDL operations via on_replace_dd_user() trigger.
+	 */
+	int64_t tx_id;
+	/**
+	 * Marker indicating that this user is deleted by some transaction.
+	 * Always false outside transactions.
+	 */
+	bool is_deleted;
 	/**
 	 * An id in privileges array to quickly find a
 	 * respective privilege.
@@ -190,6 +200,9 @@ user_find_by_name_xc(const char *name, uint32_t len)
 	return user;
 }
 
+bool
+user_check_tx_ownership(struct user *user);
+
 /** Initialize the user cache and access control subsystem. */
 void
 user_cache_init(void);

test/box/error.result

@@ -443,6 +443,7 @@ t;
  |   222: box.error.QUORUM_WAIT
  |   223: box.error.INTERFERING_PROMOTE
  |   224: box.error.RAFT_DISABLED
+ |   225: box.error.USER_BUSY
  | ...
 
 test_run:cmd("setopt delimiter ''");