FLX subscriptions are not sent to the server

[sync-by-unito]

While updating BAAS in #7042 the test "app: flx-sync basic tests" regularly hangs while waiting for a FLX subscription to be marked complete. This is reproducible locally.

A race in send_upload_message() causes subscriptions not being sent to the server (and as a side effect a notification on that subscription may never be fulfilled). This can happen when send_upload_message() is called as a result of a commit previous to the one for subscriptions. send_upload_message() then forwards the upload cursor past the version associated with the subscription and so the subscription is never uploaded to the server.

[ironage]

For reference, here are the truncated logs of the test "app: flx-sync basic tests", showing the series of events leading to a hang waiting on the fulfillment of a subscription:

DB: 4193 Thread 0x16d31b000: subscription version 3 updating state from AwaitingMark to Complete
DB: 4193 Thread 0x16d31b000: Query find all: 'version < 3', limit = -1
DB: 4193 Thread 0x16d31b000: Query found: 1, Duration: 12 us
DB: 4193 Thread 0x16d31b000: Initiate commit version: 26
DB: 4193 Thread 0x16d31b000: db state for version 26 is: 
{
"sync_internal_schemas":[{"_key":0,"schema_group_name":"flx_subscription_store","schema_version":2},{"_key":1,"schema_group_name":"pending_bootstraps","schema_version":1}]
,"flx_subscription_sets":[{"_key":3,"version":3,"state":3,"snapshot_version":21,"error":null,"subscriptions":[{"_key":3,"id":"6531b6e2fcbfea88cbae7b8c","created_at":"2023-10-19 23:08:18.980461000","updated_at":"2023-10-19 23:08:19.049395000","name":null,"object_class":"Obj","query":"name == \"bar\" "}]}]
,"metadata":[{"_key":0,"version":1}]
,"class_Obj":[{"_key":1,"_id":"6531b6e0fcbfea88cbae7b89","name":"bar","value":10}]
,"flx_pending_bootstrap":[]
}


DB: 4193 Thread 0x16d31b000: Commit of size 3648 done in 1889 us ref 12608
DB: 4193 Thread 0x16d31b000: writemutex released
DB: 4193 Thread 0x202852080: End transaction 1869
DB: 4193 Thread 0x16d31b000: Start frozen 1796: 26 ref 12608
DB: 4193 Thread 0x202852080: Tr 1709: Advance read: 21 -> 26 ref 12608
DB: 4193 Thread 0x16d31b000: End transaction 1796
DB: 4193 Thread 0x16d31b000: End transaction 1981
DB: 4193 Thread 0x202852080: Start frozen 2034: 26 ref 12608
DB: 4193 Thread 0x16d31b000: Start read 1796: 26 ref 12608
DB: 4193 Thread 0x16d31b000: End transaction 1796
DB: 4193 Thread 0x16d31b000: Start read 1796: 26 ref 12608
DB: 4193 Thread 0x16d31b000: End transaction 1796
DB: 4193 Thread 0x202852080: End transaction 2034
DB: 4193 Thread 0x202852080: acquire writemutex
DB: 4193 Thread 0x202852080: writemutex acquired
DB: 4193 Thread 0x202852080: Start write 1981: 26 ref 12608
DB: 4193 Thread 0x202852080: Initiate commit version: 27
DB: 4193 Thread 0x202852080: db state for version 27 is: 
{
"sync_internal_schemas":[{"_key":0,"schema_group_name":"flx_subscription_store","schema_version":2},{"_key":1,"schema_group_name":"pending_bootstraps","schema_version":1}]
,"flx_subscription_sets":[{"_key":3,"version":3,"state":3,"snapshot_version":21,"error":null,"subscriptions":[{"_key":3,"id":"6531b6e2fcbfea88cbae7b8c","created_at":"2023-10-19 23:08:18.980461000","updated_at":"2023-10-19 23:08:19.049395000","name":null,"object_class":"Obj","query":"name == \"bar\" "}]},{"_key":4,"version":4,"state":1,"snapshot_version":26,"error":null,"subscriptions":[]}]
,"metadata":[{"_key":0,"version":1}]
,"class_Obj":[{"_key":1,"_id":"6531b6e0fcbfea88cbae7b89","name":"bar","value":10}]
,"flx_pending_bootstrap":[]
}


DB: 4193 Thread 0x16d31b000: Start read 1796: 26 ref 12608
DB: 4193 Thread 0x202852080: Commit of size 2424 done in 932 us ref 10432
DB: 4193 Thread 0x16d31b000: Query find all: 'version > 3 and (state == 1 or state == 2) and snapshot_version > 24', limit = -1
DB: 4193 Thread 0x202852080: writemutex released
DB: 4193 Thread 0x16d31b000: Query found: 0, Duration: 31 us
DB: 4193 Thread 0x16d31b000: End transaction 1796
DB: 4193 Thread 0x202852080: Start frozen 1796: 27 ref 10432
DB: 4193 Thread 0x16d31b000: Start read 2034: 27 ref 10432
DB: 4193 Thread 0x202852080: End transaction 1796
DB: 4193 Thread 0x16d31b000: End transaction 2034
Connection[2]: Session[2]: Sending: UPLOAD(progress_client_version=27, progress_server_version=24, locked_server_version=24, num_changesets=0)
DB: 4193 Thread 0x16d31b000: Start read 2034: 27 ref 10432
DB: 4193 Thread 0x16d31b000: End transaction 2034
DB: 4193 Thread 0x16d31b000: Start read 2034: 27 ref 10432
DB: 4193 Thread 0x16d31b000: End transaction 2034
DB: 4193 Thread 0x16d31b000: Start read 1796: 27 ref 10432

// NOTE: the following log is from Session::send_message(). The method get_next_pending_version does not find 
// the just committed subscription because the immediately previous upload just ran on version 27. 
//   m_pending_flx_sub_set = get_flx_subscription_store()->get_next_pending_version(
//   m_last_sent_flx_query_version, m_upload_progress.client_version);

DB: 4193 Thread 0x16d31b000: Query find all: 'version > 3 and (state == 1 or state == 2) and snapshot_version > 26', limit = -1
DB: 4193 Thread 0x16d31b000: Query found: 0, Duration: 34 us
DB: 4193 Thread 0x16d31b000: End transaction 1796
Connection[2]: Sending: PING(timestamp=2088136071, rtt=0)
Connection[2]: Received: PONG(timestamp=2088136071)
Connection[2]: Round trip time was 2 milliseconds
Connection[2]: Will emit a ping in 54487 milliseconds

// Hang forever waiting for subscription to be fulfilled.