You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
@@ -4,187 +4,123 @@ When a batch of SUB or NSUB commands arrives from a service client, each command
4
4
5
5
## Goal
6
6
7
-
Reduce to 1 DB query per batch, using `UPDATE ... RETURNING recipient_id` to identify which queues were actually updated.
7
+
Reduce to at most 2 DB queries per batch (one for rcv associations, one for ntf associations), using `UPDATE ... RETURNING recipient_id` to identify which queues were actually updated.
8
8
9
-
`clntServiceId` is per-client (not per-queue), so all queues in a batch that need association changes share the same target value. The per-queue decision is binary: update or not.
9
+
Also fuse message pre-fetch and association batching into a single batch preparation step with a clean contract.
10
10
11
-
```haskell
12
-
typeNeedsAssocUpdate=Bool
13
-
```
14
-
15
-
## Current code
16
-
17
-
### `sharedSubscribeQueue` (Server.hs:1757-1805)
18
-
19
-
Called per SUB/NSUB command.Based on `clntServiceId`and`queueServiceId` from `QueueRec`:
Only from the `client` function's `foldrM` loop in`Server.hs` (via `processCommand`->`subscribeQueueAndDeliver`or`subscribeNotifications`).The forwarded command handler (line 2094) only processes sender commands, never SUB/NSUB.So`prepareBatch` always runs before `sharedSubscribeQueue`.
32
-
33
-
### `setQueueService` for Postgres (QueueStore/Postgres.hs:484-505)
11
+
## Contract
34
12
35
-
Per queue:
36
-
1. `withQueueRec sq` -readsQueueRecTVar under queue lock, fails if deleted
37
-
2.Checksif already set to target value - returns immediately if so
38
-
3. `assertUpdated $ withDB' ...DB.execute "UPDATE ..."` - one DB query, asserts 1 row affected
- `MaybeMessage` - pre-fetched message for SUB queues, `Nothing` for NSUBor no message
21
+
- `Maybe (EitherErrorType())` - association result. `Nothing` = no update needed. `Just (Right())` = update succeeded. `Just (Left e)` = update failed for this queue.
70
22
71
-
`assocResults` contains entries only for queues that needed an association update. Keyed by `RecipientId`. `Right ()` means the update succeeded. `Left e` means it failed.
23
+
Onemap, one lookup per queue.`processCommand` passes both values to `subscribeQueueAndDeliver`/`subscribeNotifications`->`sharedSubscribeQueue`.
72
24
73
-
### Step 2: Implement `prepareBatch` (Server.hs)
25
+
Queuesnotin the map (non-SUB/NSUB commands, failed verification) are not affected.
74
26
75
-
Replaces current `prefetchMsgs`. Does three things:
27
+
## prepareBatch implementation
76
28
77
-
1. Collects SUB queues for message pre-fetch (existing `tryPeekMsgs` logic, unchanged).
29
+
One accumulating fold over the batch, collecting three lists:
- `ntfAssocQs :: [StoreQueues]` -NSUBqueuesneeding `ntf_service_id` update (`clntServiceId/=ntfServiceId` from `NtfCreds`)
78
33
79
-
2. Classifies each SUB/NSUB queue's association need by reading `queueServiceId`from the already-loaded `QueueRec` in `VerifiedTransmission`and comparing with `clntServiceId`. Produces `NonEmpty (Either ErrorType NeedsAssocUpdate)` aligned with the batch. Error if `q_ = Nothing` for a SUB/NSUB command. `True` if the queue needs its association updated. `False` if no change needed.
34
+
Classificationreadsfrom the already-loaded `QueueRec` in `VerifiedTransmission` -no extra DB query.
80
35
81
-
3. Collects `StoreQueue`s where classification produced `Right True`. If non-empty, calls `setQueueServices` with `clntServiceId` as target and this list. Gets back `Set RecipientId` of queues that were actually updated.
36
+
Then three store calls (each skipped if its list is empty):
37
+
1. `tryPeekMsgs ms subMsgQs` -> `MapRecipientIdMessage`
4. Builds `assocResults :: Map RecipientId (Either ErrorType ())`: for each queue that needed an update (`Right True`), if its `recipientId` is in the returned set then `Right ()`, otherwise `Left AUTH`.
41
+
Then one pass to merge results into `MapRecipientId (MaybeMessage, Maybe (EitherErrorType()))`:
42
+
-For each SUB queue: `(M.lookup rId msgMap, assocResult rId rcvUpdated rcvAssocQs)`
43
+
-For each NSUB queue: `(Nothing, assocResult rId ntfUpdated ntfAssocQs)`
84
44
85
-
### Step 3: Thread `assocResults` through `processCommand` (Server.hs:1463)
45
+
Where `assocResult rId updated assocQs` =if the queue was in`assocQs` (needed update), then `Just (Right())` if`rId` is in`updated`, else `Just (LeftAUTH)`.Ifnotin`assocQs` (no update needed), `Nothing`.
No `SParty p` polymorphism.Each function knows its column.
107
59
108
-
Gains `assocResult ::Maybe (EitherErrorType())`.
60
+
### Postgres implementation
109
61
110
-
Where the queue needs a new or changed association (line 1772), currently:
111
-
```haskell
112
-
|otherwise-> runExceptT $do
113
-
ExceptT$ setQueueService (queueStore ms) q party (Just serviceId)
114
-
hasSub <-...
62
+
`setRcvQueueServices`:
63
+
```sql
64
+
UPDATE msg_queues SET rcv_service_id =?
65
+
WHERE recipient_id IN?AND deleted_at ISNULL
66
+
RETURNING recipient_id
115
67
```
116
68
117
-
Becomes:
118
-
```haskell
119
-
|otherwise->case assocResult of
120
-
Just (Left e) ->pure$Left e
121
-
_ -> runExceptT $do
122
-
hasSub <-...
69
+
`setNtfQueueServices`:
70
+
```sql
71
+
UPDATE msg_queues SET ntf_service_id = ?
72
+
WHERE recipient_id IN ? AND notifier_id IS NOT NULLAND deleted_at IS NULL
73
+
RETURNING recipient_id
123
74
```
124
75
125
-
`Just (Left e)` means the batch update failed for this queue - return the error.
126
-
`Just (Right ())` means the batch update succeeded - skip `setQueueService`, proceed with STM work.
127
-
`Nothing` cannot happen here because `prepareBatch` always runs before this code and classifies every SUB/NSUB queue.
76
+
After each batch query, for each queue in the returned set:
77
+
1. Read QueueRec TVar, update with new serviceId
78
+
2. Write store log entry
128
79
129
-
Same change where removing association (line 1787):
130
-
```haskell
131
-
Just _ ->case assocResult of
132
-
Just (Left e) ->pure$Left e
133
-
_ -> runExceptT $do
134
-
liftIO $ incSrvStat srvAssocRemoved
135
-
...
136
-
```
80
+
### STM implementation
137
81
138
-
Queues that are already associated correctly or have no service involvement have `assocResult = Nothing` (not in the map). These paths don't call `setQueueService` today, so nothing changes for them.
82
+
Loop over queues, call existing per-item logic, collect succeeded `RecipientId`s into a Set.
139
83
140
-
### Step 6: Add `setQueueServices` to `QueueStoreClass` (QueueStore/Types.hs:53)
SUB case: `M.lookup entId prepared` gives `Just (msg_, assocResult)` or `Nothing`. Pass both to `subscribeQueueAndDeliver`.
149
91
150
-
For `SRecipientService`:
151
-
```sql
152
-
UPDATE msg_queues SET rcv_service_id =?
153
-
WHERE recipient_id IN?AND deleted_at ISNULL
154
-
RETURNING recipient_id
155
-
```
92
+
NSUB case: `M.lookup entId prepared` gives `Just (Nothing, assocResult)` or `Nothing`. Pass `assocResult` to `subscribeNotifications`.
156
93
157
-
For `SNotifierService`:
158
-
```sql
159
-
UPDATE msg_queues SET ntf_service_id = ?
160
-
WHERE recipient_id IN ? AND notifier_id IS NOT NULLAND deleted_at IS NULL
161
-
RETURNING recipient_id
162
-
```
94
+
Forwarded commands: pass `M.empty`.
163
95
164
-
Build `Set RecipientId` from RETURNING rows.
96
+
### subscribeQueueAndDeliver
165
97
166
-
After the batch query, for each queue whose `recipientId` is in the returned set:
167
-
1. Read `QueueRec` from TVar, update with new `serviceId` (same as `updateQueueRec` at line 502-504)
168
-
2. Write store log entry (same as `withLog` at line 505)
98
+
Takes `Maybe Message` and `Maybe (Either ErrorType ())` as before. No change in how it uses them.
169
99
170
-
Queues not in the returned set are not updated (deleted between verification and UPDATE). The caller sees them absent from the set and produces `Left AUTH`.
100
+
### sharedSubscribeQueue
171
101
172
-
No per-queue lock needed: the batch UPDATE is a single SQL statement (Postgres handles row-level locking internally), and SUB/NSUB processing is single-threaded per connected client.
102
+
Takes `Maybe (Either ErrorType ())`. On paths needing association update:
103
+
-`Just (Left e)` -> return error
104
+
-`Just (Right ())` -> skip `setQueueService`, proceed with STM work
105
+
-`Nothing` -> no update needed, proceed with existing logic
1. Run existing `setService` STM logic from `setQueueService` (line 319-334): read QueueRec, update TVar, update per-service queue sets
178
-
2. If succeeded, add `recipientId` to result set
179
-
3. Write store log entry
109
+
1. Define the `prepareBatch` contract and thread one map through `processCommand` -> `subscribeQueueAndDeliver` / `subscribeNotifications` -> `sharedSubscribeQueue` (Server.hs)
110
+
2. Implement `prepareBatch` with the fold, three calls, and merge (Server.hs)
111
+
3. Add `setRcvQueueServices` and `setNtfQueueServices` to `QueueStoreClass` (Types.hs)
112
+
4. Implement for Postgres with batch `UPDATE ... RETURNING` (Postgres.hs)
113
+
5. Implement for STM as loop (STM.hs)
114
+
6. Implement for Journal as delegation (Journal.hs)
180
115
181
-
Return accumulated `Set RecipientId`.
116
+
At step 2, store functions can initially be stubs returning empty sets. Steps 3-6 fill in the real implementations.
182
117
183
118
## Files changed
184
119
185
120
| File | Change |
186
121
|---|---|
187
-
|`src/Simplex/Messaging/Server.hs`| Rename `prefetchMsgs` to `prepareBatch` adding classification and `setQueueServices` call. Thread `assocResults` through `processCommand` -> `subscribeQueueAndDeliver` / `subscribeNotifications` -> `sharedSubscribeQueue`. Replace `setQueueService` calls with `assocResult` check. |
188
-
|`src/Simplex/Messaging/Server/QueueStore/Types.hs`| Add `setQueueServices` to `QueueStoreClass`|
189
-
|`src/Simplex/Messaging/Server/QueueStore/Postgres.hs`| Implement `setQueueServices` with batch `UPDATE ... RETURNING` + per-item TVar and store log updates |
190
-
|`src/Simplex/Messaging/Server/QueueStore/STM.hs`| Implement `setQueueServices` as loop over existing STM logic |
122
+
|`src/Simplex/Messaging/Server.hs`|`prepareBatch` with fold + merge; one map parameter through `processCommand` -> `subscribeQueueAndDeliver` / `subscribeNotifications` -> `sharedSubscribeQueue`|
123
+
|`src/Simplex/Messaging/Server/QueueStore/Types.hs`| Add `setRcvQueueServices`, `setNtfQueueServices` to `QueueStoreClass`|
0 commit comments