From f26d9fc7fd103841fb27661b5d72f74a1ea9daff Mon Sep 17 00:00:00 2001 From: Fernando Takagi Date: Wed, 25 Oct 2023 11:06:33 -0300 Subject: [PATCH 1/6] Ignore insert if it exists. --- server/core_storage.go | 1 + 1 file changed, 1 insertion(+) diff --git a/server/core_storage.go b/server/core_storage.go index 3240318cd..b5e9a62c9 100644 --- a/server/core_storage.go +++ b/server/core_storage.go @@ -699,6 +699,7 @@ func storagePrepBatch(batch *pgx.Batch, authoritativeWrite bool, op *StorageOpWr query = ` INSERT INTO storage (collection, key, user_id, value, version, read, write, create_time, update_time) VALUES ($1, $2, $3, $4, $5, $6, $7, now(), now()) + ON CONFLICT (collection, key, user_id) DO NOTHING RETURNING read, write, version, create_time, update_time` // Outcomes: From a5cbeba33492ac3c009979537280d3a7e5512ebf Mon Sep 17 00:00:00 2001 From: Fernando Takagi Date: Wed, 25 Oct 2023 11:22:38 -0300 Subject: [PATCH 2/6] Update core_storage.go --- server/core_storage.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/server/core_storage.go b/server/core_storage.go index b5e9a62c9..955ad3eb3 100644 --- a/server/core_storage.go +++ b/server/core_storage.go @@ -699,11 +699,11 @@ func storagePrepBatch(batch *pgx.Batch, authoritativeWrite bool, op *StorageOpWr query = ` INSERT INTO storage (collection, key, user_id, value, version, read, write, create_time, update_time) VALUES ($1, $2, $3, $4, $5, $6, $7, now(), now()) - ON CONFLICT (collection, key, user_id) DO NOTHING + ON CONFLICT (collection, key, user_id) DO NOTHING RETURNING read, write, version, create_time, update_time` // Outcomes: - // - NoRows - insert failed due to constraint violation (concurrent insert) + // - NoRows - insert was ignored due to constraint violation (concurrent insert or non-OCC existing row) } batch.Queue(query, params...) From b01f63307bf159d4d5d2cbc1e7390b6c6f48e2ee Mon Sep 17 00:00:00 2001 From: Fernando Takagi Date: Wed, 25 Oct 2023 18:28:52 -0300 Subject: [PATCH 3/6] return success if writing same data --- server/core_storage.go | 17 ++++++++++++----- 1 file changed, 12 insertions(+), 5 deletions(-) diff --git a/server/core_storage.go b/server/core_storage.go index 955ad3eb3..4aeae06dc 100644 --- a/server/core_storage.go +++ b/server/core_storage.go @@ -697,13 +697,20 @@ func storagePrepBatch(batch *pgx.Batch, authoritativeWrite bool, op *StorageOpWr // OCC if-not-exists, and all other non-OCC cases. // Existing permission checks are not applicable for new storage objects. query = ` - INSERT INTO storage (collection, key, user_id, value, version, read, write, create_time, update_time) - VALUES ($1, $2, $3, $4, $5, $6, $7, now(), now()) - ON CONFLICT (collection, key, user_id) DO NOTHING - RETURNING read, write, version, create_time, update_time` + WITH ins AS ( + INSERT INTO storage (collection, key, user_id, value, version, read, write, create_time, update_time) + VALUES ($1, $2, $3, $4, $5, $6, $7, now(), now()) + ON CONFLICT (collection, key, user_id) DO NOTHING + RETURNING read, write, version, create_time, update_time + ) + (SELECT read, write, version, create_time, update_time FROM ins) + UNION ALL + (SELECT read, write, version, create_time, update_time FROM storage WHERE collection = $1 AND key = $2 AND user_id = $3 AND version = $5 AND read = $6 AND write = $7) + LIMIT 1` // Outcomes: - // - NoRows - insert was ignored due to constraint violation (concurrent insert or non-OCC existing row) + // - Row is returned if object was new or if object already existed but had the same values being inserted + // - NoRows - insert was ignored due to constraint violation (concurrent insert or non-OCC existing row), and values being inserted were different from existing row } batch.Queue(query, params...) From bde48e5adaa664f1ab531aed1b5ac09fbbdc6097 Mon Sep 17 00:00:00 2001 From: Fernando Takagi Date: Thu, 26 Oct 2023 16:42:13 -0300 Subject: [PATCH 4/6] Change to a concurrent safe query. --- server/core_storage.go | 20 ++++++++------------ 1 file changed, 8 insertions(+), 12 deletions(-) diff --git a/server/core_storage.go b/server/core_storage.go index 4aeae06dc..ee5ff0c1d 100644 --- a/server/core_storage.go +++ b/server/core_storage.go @@ -697,20 +697,16 @@ func storagePrepBatch(batch *pgx.Batch, authoritativeWrite bool, op *StorageOpWr // OCC if-not-exists, and all other non-OCC cases. // Existing permission checks are not applicable for new storage objects. query = ` - WITH ins AS ( - INSERT INTO storage (collection, key, user_id, value, version, read, write, create_time, update_time) - VALUES ($1, $2, $3, $4, $5, $6, $7, now(), now()) - ON CONFLICT (collection, key, user_id) DO NOTHING - RETURNING read, write, version, create_time, update_time - ) - (SELECT read, write, version, create_time, update_time FROM ins) - UNION ALL - (SELECT read, write, version, create_time, update_time FROM storage WHERE collection = $1 AND key = $2 AND user_id = $3 AND version = $5 AND read = $6 AND write = $7) - LIMIT 1` + INSERT INTO storage (collection, key, user_id, value, version, read, write, create_time, update_time) + SELECT $1, $2, $3, $4, $5, $6, $7, now(), now() + WHERE NOT EXISTS(SELECT 1 FROM storage WHERE collection = $1 AND key = $2 AND user_id = $3) -- avoids ON CONFLICT hitting pre-existing object and immediately cancels insert + ON CONFLICT (collection, key, user_id) DO UPDATE + SET version = excluded.version WHERE storage.version = excluded.version AND storage.read = excluded.read AND storage.write = excluded.write -- needed to return a row on concurrent write of same object + RETURNING read, write, version, create_time, update_time` // Outcomes: - // - Row is returned if object was new or if object already existed but had the same values being inserted - // - NoRows - insert was ignored due to constraint violation (concurrent insert or non-OCC existing row), and values being inserted were different from existing row + // - Row is returned if object was new or if concurrent writer wrote object with the same values (ON CONFLICT UPDATE needed) + // - NoRows - insert was ignored due to constraint violation (non-OCC already existing row), or concurrent write wrote object with different values (UPDATE did not happen) } batch.Queue(query, params...) From 1f13eed7bbece3ab5f3b29d4246854b04dd78b63 Mon Sep 17 00:00:00 2001 From: Fernando Takagi Date: Thu, 26 Oct 2023 16:52:00 -0300 Subject: [PATCH 5/6] Update core_storage.go --- server/core_storage.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/server/core_storage.go b/server/core_storage.go index ee5ff0c1d..19ae548c3 100644 --- a/server/core_storage.go +++ b/server/core_storage.go @@ -699,7 +699,7 @@ func storagePrepBatch(batch *pgx.Batch, authoritativeWrite bool, op *StorageOpWr query = ` INSERT INTO storage (collection, key, user_id, value, version, read, write, create_time, update_time) SELECT $1, $2, $3, $4, $5, $6, $7, now(), now() - WHERE NOT EXISTS(SELECT 1 FROM storage WHERE collection = $1 AND key = $2 AND user_id = $3) -- avoids ON CONFLICT hitting pre-existing object and immediately cancels insert + WHERE NOT EXISTS(SELECT 1 FROM storage WHERE collection = $1 AND key = $2 AND user_id = $3) -- avoids ON CONFLICT hitting pre-existing object, and immediately cancels insert ON CONFLICT (collection, key, user_id) DO UPDATE SET version = excluded.version WHERE storage.version = excluded.version AND storage.read = excluded.read AND storage.write = excluded.write -- needed to return a row on concurrent write of same object RETURNING read, write, version, create_time, update_time` From 3decf8759494f384e7ca7f0a09a992958ce71390 Mon Sep 17 00:00:00 2001 From: Fernando Takagi Date: Thu, 26 Oct 2023 18:24:43 -0300 Subject: [PATCH 6/6] Update core_storage.go --- server/core_storage.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/server/core_storage.go b/server/core_storage.go index 19ae548c3..35e40ccf3 100644 --- a/server/core_storage.go +++ b/server/core_storage.go @@ -701,7 +701,8 @@ func storagePrepBatch(batch *pgx.Batch, authoritativeWrite bool, op *StorageOpWr SELECT $1, $2, $3, $4, $5, $6, $7, now(), now() WHERE NOT EXISTS(SELECT 1 FROM storage WHERE collection = $1 AND key = $2 AND user_id = $3) -- avoids ON CONFLICT hitting pre-existing object, and immediately cancels insert ON CONFLICT (collection, key, user_id) DO UPDATE - SET version = excluded.version WHERE storage.version = excluded.version AND storage.read = excluded.read AND storage.write = excluded.write -- needed to return a row on concurrent write of same object + SET version = excluded.version + WHERE storage.version = excluded.version AND storage.read = excluded.read AND storage.write = excluded.write -- needed to return a row on concurrent write of same object RETURNING read, write, version, create_time, update_time` // Outcomes: