Idempotent persistent-RPC creation under re-delivery - #37
Open
ShvaykaD wants to merge 15 commits into
Open
Conversation
Both RPC repositories wrapped the serialized JSON in replaceNullChars before binding it. That scrub cannot match anything here: it strips the raw U+0000 character, while JacksonUtil.toString escapes control chars, so the serialized string only ever carries the six-character backslash-u-0000 escape. Verified for a NUL in a value, in a field name, and put into a node programmatically - all three round-trip unchanged through it. It was also new behaviour for the rpc table. Before the JDBC write paths existed, RPCs went through Hibernate and JsonConverter, which is exactly JacksonUtil.toString with no scrub, so this only ever added a Matcher allocation and a full-string regex pass per JSON column per row on the create/status-update hot path. The columns are varchar, not jsonb, so the escaped form stores as it always has. JacksonUtil.toString already returns null for a null argument, so the explicit null branch went with it and response/additional_info still bind SQL NULL.
…aining it Trim the commentary added with the insert-if-absent path down to what the code cannot say itself: - RpcInsertRepository: both blocks. The INSERT_IF_ABSENT block repeated the constant name, the ON CONFLICT DO NOTHING clause below it and the RpcDao javadoc; the save() block was naming archaeology whose one fact (it returns a result) is the method signature. - RpcDao.createIfAbsent: drop "Insert-if-absent" (the method name) and the note about not being called save(...). The contract sentence and @return stay - a caller cannot read those off the signature. - DeviceActorMessageProcessor: drop the trailing "no-op if the row already exists" (the block two lines down already says the reply is not gated on the create result) and shorten the duplicate-delivery block, whose first and last sentences restated the if-condition, the log line and the call below it. What survives in the duplicate-delivery block is the part that is not visible locally: which row owns delivery instead, and that re-sending would execute the command twice on the device.
…fAbsent createIfAbsent copied the createdTime == 0 backfill from JpaAbstractDao.save, but it cannot fire. The single production caller is DeviceActorMessageProcessor.createRpc, and buildRpc sets createdTime unconditionally from the value captured at the top of the send path - which is exactly what 2757172 established, when createdTime became a value captured once at create and threaded through ToDeviceRpcRequestMetadata so the update-path notifications stop reporting the update moment. The copy was faithful to the branch that used to run (an RpcEntity always carries the rpcId, so RPCs always took the else branch of the isNew check), but that branch never fired either, for the same reason. Half of it was doubly dead: the v1 ternary, given RPC ids are v4 from REST and the gateway. TenantServiceTest.createAndSaveRpcFor was the only caller anywhere leaning on the backfill. Set createdTime there explicitly - without it the fixture silently writes created_time = 0 and the test still passes, since it only asserts the rows disappear with the tenant. Note the invariant is now caller-enforced: created_time is NOT NULL but 0 is a legal value, and a 0 row would sort ahead of every real one in loadInFlightRpcs on actor init. JpaRpcDaoTest round-trips the column.
…JPA merge Review follow-ups on the create path. No behavior change for callers, except that the JPA merge now throws instead of silently upserting - it had no callers left. - Name the operation once. TbRpcService.create -> RpcService.create -> RpcDao.createIfAbsent meant three names for one call, so `if (rpcService.create(rpc))` read as "did the save succeed" rather than "did I win the insert" - exactly the distinction the actor's duplicate-skip hangs on. All layers are createIfAbsent now, including the actor's private helper and RpcInsertRepository.save -> insertIfAbsent. - Drop the unused tenantId from RpcDao.createIfAbsent. The only caller passed rpc.getTenantId(), a value derived from the other argument. Dao.save(TenantId, T) sets a precedent for carrying it, but the closest sibling on this interface, updateAsync(Rpc), had already dropped it; the two write methods are symmetric again. TbRpcService keeps its tenantId - it genuinely needs one for the rule engine notification. - Decouple RpcInsertRepository from AbstractInsertRepository. That base lives in the timeseries sqlts.insert package and bought one thing here, the autowired jdbcTemplate: the inherited transactionTemplate is unused, and replaceNullChars went dead for the whole rpc package when 6eed4726 dropped the scrub. A plain @repository with an injected JdbcTemplate states what the class actually needs. The statement itself stays a literal, matching RpcUpdateRepository next door and every other native-SQL repository in dao. - Flatten the duplicate-RPC guard. The QUEUED persist - the mandatory persist-before-send write - sat inside a negated `else if` condition, which is easy to read past, and the else was redundant since the timeout<=0 branch always returns. The write is a statement again, with the result in a named local. - Log ids, not the whole row. Both write paths traced the entire Rpc, request payload included. tenantId, deviceId, rpcId and status match the style the rest of BaseRpcService already uses. - Make the unguarded merge unreachable. Nothing calls rpcDao.save/saveAndFlush for RPCs any more, but Dao<T> still exposes both, so the upsert this whole path replaces was one call away from coming back. Both throw now. Both need overriding: JpaAbstractDao.saveAndFlush delegates to a private save(tenantId, domain, flush) overload, not to the public save, so covering one would have left the other live. Checked the generic Dao<?> consumers first - JpaRpcDao is in EntityDaoRegistry, but uniquifyEntityName, EdqsSyncService, TenantEntitiesDeletionTaskProcessor, BaseEdgeProcessor and DefaultExportableEntitiesService only read. Tests: - Route the DAO fixtures through the production create path. The guard/update tests seeded rows with saveAndFlush, so they asserted against rows shaped by a code path nothing else takes. - Drop requestIdRoundTripsThroughSaveAndLoad and onewayRoundTripsThroughSaveAndLoad. Both assert a column survives a round-trip, but through Hibernate's mapping - the wrong path for the risk that now exists, which is the hand-written INSERT column list. createIfAbsentInsertsWhenRowMissing already covers both columns, through the path production uses. - Cover the closed merge: saveIsUnsupportedSoTheUnguardedMergeStaysUnreachable asserts both entry points throw and that neither wrote a row. - Give deleteOutdated its own tenants. Its Long.MAX_VALUE case deletes every row of the tenant it is given, and AbstractJpaDaoTest has no per-test rollback while a dozen tests in this class seed SYS_TENANT_ID rows and leave them behind - so the exact counts held only as long as JUnit 4's hash ordering kept this method first, which adding any test method can reshuffle. Two random tenant ids make them order-independent. - Assert the rpcId reply on the ordinary persisted path too. Only the duplicate and expired-on-arrival paths checked it, so nothing would have failed if the sendRpcIdResponse extraction had dropped the reply on the most common path. The captor block is now an assertRpcIdReplied helper shared by all four. - Fold the duplicated nine-argument ToDeviceRpcRequest construction into persistedRequest(rpcId, expirationTime, oneway), which the two-arg overload, expiredRequest and the row(...) fixture all delegate to, so the constructor appears exactly once in the file.
…re save-era names More review follow-ups. No behavior change. - RpcUpdateRepository no longer extends AbstractInsertRepository. After the null-char scrub went away the base supplied nothing but two @Autowired templates, and it is an odd fit besides: it lives in the timeseries sqlts.insert package while this class only issues UPDATEs. Constructor-injecting JdbcTemplate and TransactionTemplate makes both RPC write repositories read the same way, and leaves neither of them depending on the timeseries package. - Retire the names left over from the save path. JpaRpcDaoTest.saveRpc -> seedRpc (and its toSave local -> toCreate), TenantServiceTest.createAndSaveRpcFor -> createRpcFor: neither saves anything any more, so the names described an operation that no longer exists. The toSave local in saveIsUnsupportedSoTheUnguardedMergeStaysUnreachable stays as it is - there it really is the argument to save(). - Drop the "D6:" tag from expiredOnArrivalRpcReturnsRpcIdToCaller. It points at a decision list outside the repo that a future reader cannot resolve, it was the only such marker in the file, and the sentence after it already carries the meaning.
… continuation The synchronous insert-if-absent create blocked a device-dispatcher thread for a full commit round trip. Under load that saturated storage on commit count - 91% of the provisioned 125 MB/s, 300 ms write latency - and capped throughput at pool_size / insert_latency, with 96 of 96 sampled dispatcher threads parked in Net.poll inside the insert. Route the create through the batched write queue and split processRpcRequest into two actor mailbox turns: - Turn 1 (arrival) enqueues the insert and registers the pending entry right away, flagged not-yet-durable. Nothing is sent and no rpcId is replied. - processRpcPersistResult (turn 2) arrives as a local self-tell when the flush settles, clears the flag, replies the rpcId, and sends. Ordering is established at arrival, not at flush. toDeviceRpcPendingMap is ordered by insertion and the send paths walk it front-to-back, so registering on the arrival thread keeps queue position equal to arrival order no matter which stripe worker flushes first or how batch_sort reorders rows within a batch. getFirstRpc now stops at a head whose row is not durable instead of stepping over it - without that, a later command whose insert flushed first would reach the device first. Persist-before-send becomes structural: the send path cannot observe an entry whose insert has not confirmed. Creates and status updates share one rpcId-striped queue, applied inserts-before- updates in a single transaction, so an update coalesced into the same flush as its create can never be applied first and strand a QUEUED row. Per-row affected counts decide each result, so an insert that conflicted reports false and is recognised as a duplicate rather than sent to the device a second time. The continuation runs on a dedicated pool (sql.rpc.continuation_threads, default 3) rather than the notification stripes: it is on the command-delivery path, while those stripes carry request JSON serialization plus a rule-engine publish. Sharing them would let a queue stall or rule-engine backpressure hold up device sends. The RPC_QUEUED stripe task is enqueued before the actor is resumed, in one callback, so notification order stays deterministic rather than depending on Guava listener ordering. Turn 2 also advances the sequential queue when a create fails or turns out to be a duplicate - such an entry held the head from arrival, so dropping it without advancing would stall every later command for that device until expiry. isSendNewRpcAvailable stays for the non-persistent path, which keeps its single arrival turn and is unchanged. AppActor and TenantActor are untouched: the continuation is told straight to the owning actor through the TbActorCtx captured in turn 1, the same way that request's expiry timeout is delivered. Per-RPC dispatch latency is now bounded by sql.rpc.batch_max_delay (50 ms) instead of a commit round trip, against a 120 s expiry.
…ne repository The synchronous insert-if-absent create was kept only so reverting to blocking persist-before-send would be a one-method change. That revert is no longer on the table - the batched create with a persist continuation is the intended behaviour - and no production code called any of the chain: TbRpcService.createIfAbsent -> RpcService.createIfAbsent -> RpcDao.createIfAbsent -> RpcInsertRepository.insertIfAbsent. Removing it strands the single-row insert, which would leave RpcInsertRepository and RpcUpdateRepository each holding one statement plus a near-identical BatchPreparedStatementSetter, both reachable only through RpcWriteRepository. So fold all three into RpcWriteRepository: both statements, one shared batch() helper behind a ColumnBinder, one transaction. This is the shape the original batching work had before the split, minus its two defects - the create is ON CONFLICT DO NOTHING rather than an upsert, and each insert's own affected-row count decides its result instead of inserts being reported as always persisted. Deliberately still not built on AbstractVersionedInsertRepository: that base updates first and inserts any row whose update matched nothing, which would resurrect an RPC deleted in the meantime, and it reports version numbers where this path needs a per-row boolean. The javadoc now records that. Also drops collateral the removal would otherwise have left behind: the UNSUPPORTED_MERGE message named createIfAbsent(Rpc), the createIfAbsentAsync javadoc had a @link to the deleted method, and JpaRpcDao kept an unused repository field. The empty-batch guard moves into batch(), so write() no longer carries two isEmpty ternaries, and both binders are static so neither captures the enclosing instance. Tests: JpaRpcDaoTest seeds through a create(Rpc) helper so each of the 19 call sites stays one line. Three deletions, all coverage-neutral rather than lost coverage - two TbRpcServiceTest cases for the removed synchronous path are covered by the continuation tests (INSERTED / DUPLICATE / FAILED), and createIfAbsentAsyncInsertsThenReportsADuplicateOnTheSecondCall was strictly weaker than createIfAbsentIsNoOpWhenRowExists, which asserts the same true-then-false plus that the existing row is untouched. The two batch-insert cases now drive RpcWriteRepository.write instead of the deleted repository. RpcService.createIfAbsent(Rpc) is a common/dao-api removal; safe because the feature has not shipped in a release yet.
…ination sql.rpc.callback_threads and the new sql.rpc.continuation_threads were effectively synonyms - a callback is a continuation - so the names did nothing to tell the two pools apart and an operator had to read the comments to know which was which. Both are genuinely post-persist callbacks; what differs is where each one delivers. Name them for that instead: - sql.rpc.rule_engine_callback_threads - publishes RPC lifecycle events to the rule engine, striped by rpcId to keep per-command order - sql.rpc.device_actor_callback_threads - resumes the device actor once its create is durable, releasing the command for delivery RULE_ENGINE is spelled out rather than abbreviated to RE, matching the eight existing env vars that use it; the codebase has no RE_ prefix anywhere. DEVICE_ACTOR rather than CORE because TbRpcService is a @TbCoreComponent, so both pools already run in tb-core and CORE would not distinguish them. Fields, thread names (rpc-rule-engine-callback-N, rpc-device-actor-callback) and the validation messages follow. Defaults stay at 3 for both. No behaviour change; the whole sql.rpc.* block is unreleased, so no migration is needed.
The batched-create change carried 24% comment lines, much of it narration of the statement underneath or the same rationale repeated in three files. Removed: - per-constant javadoc on RpcPersistResult, the RpcPersistResultActorMsg routing paragraph and the RpcWrite one-queue rationale. Sibling actor messages carry no javadoc, RpcStatus has no per-constant javadoc, and sibling records are one-liners; each of these also duplicated a comment at the place that enforces the behaviour. - the sendToSubscriptions header, which restated the method name. - test comments narrating fixture setup that the next two lines already show. Trimmed, keeping only what a future edit could silently break: - the head-of-line persisted filter and why stepping over an unconfirmed entry reorders commands. - isEligibleHead existing because getFirstRpc has a scheduling side effect. - releasePendingRpc advancing the sequential queue, or later commands stall. - the self-tell precedent and what a dropped tell costs. - inserts-before-updates, and why AbstractVersionedInsertRepository must not be reused here. - why the two callback pools are separate. Rationale for rejected alternatives moved out of the code and left in the commit messages and PR description, where it already was. Comment lines added by this branch drop from 209 to 77. No code change - the diff touches comments only.
sendPendingRequests filtered only on undelivered for the non-sequential branches, so a device subscribing to RPC - or sending SendPendingRPC - in the window between the create being enqueued and its batch flushing was handed the command before its row existed. That both broke persist-before-send and sent the command twice, since the entry's own persist result goes on to send it again. Reachable on the default BURST strategy any time a device reconnects while RPCs are being dispatched. Both branches now skip an entry until its insert has confirmed, which is safe for BURST because that strategy has no ordering contract and the entry's own persist result will send it. The sequential branch was already covered by getFirstRpc. Also reverts two unintended behaviour changes found auditing the same paths: - The expired-at-arrival reply is unconditional again. The original explicitly did not gate it on the create result - the reply carries only the id and completion is remove-once - so gating it on a failed insert made the caller wait out the core timeout instead of reading the row. - processPendingRpc no longer sets sent. It was not needed: a first send routes through the persist result, which sets it. Setting it here also changed the timeout error from NO_ACTIVE_CONNECTION to TIMEOUT for entries delivered by the reload and resubscribe paths, which the original never did. Arguably the more accurate code, but it is a separate change and does not belong here. The rest of the RPC paths are unchanged: processServerSideRpcTimeout, processRpcResponses, processRpcResponsesFromEdge, processRemoveRpc, restorePendingRpc, loadInFlightRpcs, scheduleAwaitRpcResponseFuture, isSendNewRpcAvailable, sendNextPendingRequest, buildRpc and saveRpcRequestToEdgeQueue are byte-identical to before the batching work, and processPendingRpc now differs only by using the shared proto builder, whose eight fields match the inline one.
status appears in the predicate of idx_rpc_in_flight, and Postgres treats a column named in a partial index predicate as indexed. The RPC lifecycle is a sequence of status transitions, so every one of them was disqualified from a heap-only update - 0 of 43.9M measured - and each wrote a new row version plus entries in all three indexes on rpc, roughly 4 row versions and 12 index entries per RPC where 1 and 3 would do. Dead versions could then only be reclaimed by vacuum passes walking every index, which at 100K gateways on a 1-minute RPC cadence lost by about 1.8x per pass and left 55% of the table dead before throughput collapsed. The index existed so the device-actor reload stays proportional to outstanding RPCs rather than to history. Without it the reload is an index scan on idx_rpc_tenant_id_device_id with the status test applied as a filter, so its cost is proportional to rows per device - the per-device RPC rate times the retention window, which was 31 rows at most in the measured fleet and stays bounded in any deployment that sets an RPC TTL. Nothing else used it: every other rpc query keys on (tenant_id, device_id), the primary key, or created_time. The reload query itself is unchanged, so the planner falls back on its own, and paging stays deterministic because DaoUtil.toPageable appends ORDER BY id. Measured after the change: 52.5% HOT across 82.5M updates, dead tuples at 0.0%, and 27,496,000 RPCs delivered with none expired.
The batching work added four durability gates - the sendPendingRequests filter, the two getFirstRpc filters and isEligibleHead - all reading ToDeviceRpcRequestMetadata .persisted. Non-persistent RPC is unaffected by construction: the field defaults to true, the only call site that sets it false sits inside if (request.isPersisted()), and the other two register overloads hard-code true. But nothing proved it, because every request the test suite built was persisted=true. Five tests over the non-persistent path: - sent in the arrival turn, with no create enqueued and no update issued - one-way completes on send and is never left pending - two-way stays pending after being sent - expired on arrival is dropped silently: no row, no reply, nothing registered - a pending non-persistent RPC is still delivered on subscribe The last one guards the filter added in 315254e, which is the change that could plausibly have held a non-persistent RPC back. Verified non-vacuous by inverting the filter's persisted test, which fails it with "Wanted but not invoked".
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Problem
The persistent-RPC status-update path is guarded (
UPDATE … WHERE id = ? AND status = ANY(<allowed-from>)), but the create path is not: theRpcis built with a preset id, soJpaAbstractDao.savetakes theisNew == falsebranch — a JPA merge, i.e. an unconditional upsert.So a re-delivered command (a rule engine partition re-consumed before it was acknowledged, or a retry processing strategy) clobbers the existing row back to
QUEUEDwith a freshrequestId, registers a second pending entry under a newrpcSeq, and re-sends to the device. Two in-flight attempts against one row, a non-deterministic final status, and a duplicate delivery. The guarded UPDATE can't help — the damage is done by the create.Second problem, found by load-testing this branch. The create was also synchronous — one un-batched commit on the device-dispatcher thread before
sendToTransport. At ~8,333 persistent RPC/s that saturated RDS storage on commit count, not row volume:WriteThroughputreached 91% of the provisioned 125 MB/s,WriteLatencywent 4 ms → 300 ms andDiskQueueDepth1 → 36, while CPU stayed under 40% and IOPS at half of 3,000. Thread dumps showed 96 of 96 sampled dispatcher threads parked inNet.pollinside the RPC insert, so throughput was capped atpool_size / insert_latency— raising the dispatcher pool from 10 to 24 made it worse. And becauseTbSendRPCRequestNodeonly acks its message when therpcIdreply arrives, the stall propagated upstream: the rule engine couldn't ack packs, so it couldn't produce, and RpcCalls consumption collapsed from ~6,600/s to 678/s.Goal
Make a re-delivered persistent RPC a no-op, so at-least-once delivery is safe on the create path.
Then: stop the create blocking a dispatcher thread and stop paying one commit per RPC — without giving up persist-before-send, arrival ordering, or "no send on a failed insert".
Approach
RpcInsertRepository.insertIfAbsent(RpcEntity)doingINSERT … ON CONFLICT (id) DO NOTHING, returning whether a row was inserted. One statement replaces the merge'sSELECT+INSERT/UPDATEpair.RpcService.save(Rpc)→boolean createIfAbsent(Rpc), so the unguarded upsert is no longer reachable from the service API.RpcDaogainscreateIfAbsent(Rpc)— the tenant comes off theRpcitself, matching the siblingupdateAsync(Rpc).TbRpcService.createIfAbsentreturnsinsertedand notifies the rule engine only when it actually inserted, so a duplicate emits no spuriousRPC_QUEUED.processRpcRequest— on a collision, skip the device send andregisterPendingRpcRequestand return. The existing row's own lifecycle (live pending entry, or the reload on actor init) owns delivery.JpaRpcDao.save/saveAndFlushnow throwUnsupportedOperationException. With no callers left, the merge was still one call away from coming back viaDao<T>. Both need overriding:JpaAbstractDao.saveAndFlushdelegates to a privatesave(tenantId, domain, flush)overload rather than to the publicsave, so covering one would leave the other live. Enforcement is runtime-only — making it a compile error would mean splittingDao<T>for every entity type. The genericDao<?>consumers that can reach this DAO throughEntityDaoRegistry(uniquifyEntityName,EdqsSyncService,TenantEntitiesDeletionTaskProcessor,BaseEdgeProcessor,DefaultExportableEntitiesService) all only read.The create guard is deliberately stricter than the update guard and the two must stay separate:
RpcStatus.QUEUED.getAllowedFromStatuses()is{SENT, TIMEOUT}, so a re-queue is a legal update. Only create requires plain absence.No proto / wire / transport / device change.
Behavior change: every persisted RPC now returns its
rpcId, including one that arrived already expired. That branch previously wrote theEXPIREDrow and returned silently, so the caller waited out the core's safety net for an opaqueTIMEOUTand never learned the id. Bounded by the same 1 s grace window — past it the reply is discarded as stale, as before.Caveat: this is dedup-by-id, so it only applies when the re-delivered command carries the same id.
TbSendRPCRequestNodemintsUuids.timeBased()when the metadata has norequestUUID, and never writes it back — a rule chain relying on that fallback is not re-delivery-safe. Documenting it innodeDetailsbelongs to #36.Batched create with non-blocking persist-before-send (later commits)
rpcId-striped batch queue as op-taggedRpcWrites, andRpcWriteRepositoryapplies all inserts before all updates in one transaction — so an update coalesced into the same flush as its create can't be applied first and strand aQUEUEDrow. Per-row affected counts survive batching, so a conflicted insert still reportsfalse: the duplicate signal above is unchanged.processRpcRequestsplits into two mailbox turns. Turn 1 enqueues the insert and registers the pending entry immediately, flagged not-yet-durable; it sends nothing and replies nothing.processRpcPersistResult(turn 2) arrives as a local self-tell through theTbActorCtxcaptured in turn 1 — the path this RPC's expiry timeout already uses — then clears the flag, replies therpcId, and sends.toDeviceRpcPendingMapis ordered by insertion and the send paths walk it front-to-back, so registering on the arrival thread keeps queue position equal to arrival order no matter which stripe worker flushes first.getFirstRpcnow stops at a head whose row isn't durable instead of stepping over it; without that, a later command whose insert flushed first would reach the device first. Persist-before-send becomes structural rather than procedural — the send path cannot observe an entry whose insert hasn't confirmed.sql.rpc.rule_engine_callback_threads(publishes lifecycle events, striped by rpcId) and the newsql.rpc.device_actor_callback_threads(resumes the actor, releasing the command for delivery), both default 3. Deliberately separate: the second sits on the command-delivery path, while the first carries request-JSON serialization plus a rule-engine publish, so sharing one pool would let a queue stall or RE backpressure hold up device sends. Per-task work on the delivery pool is a single enqueue, so three threads stay idle at the measured load.isSendNewRpcAvailablestays for the non-persistent path, which keeps its single arrival turn and is unchanged.AppActor/TenantActorare untouched.Supersedes two specifics above:
RpcInsertRepositoryandRpcUpdateRepositoryare folded into oneRpcWriteRepository, and the synchronousRpcService.createIfAbsent(Rpc)is removed — it had no production caller once the batched path landed. That last one is acommon/dao-apiremoval, safe because the feature hasn't shipped in a release yet.Latency: dispatch now waits up to
sql.rpc.batch_max_delay(50 ms) rather than a commit round trip, against a 120 s expiry.Tests
JpaRpcDaoTestcovers the DB-level semantics against real Postgres: the insert round-trips every bound column, a duplicate leavesrequest_id/created_time/statusuntouched, a duplicate against a row already driven toSUCCESSFULneither resurrects it nor drops the response, and bothsaveentry points throw without writing a row. Its fixtures seed throughcreateIfAbsentrather than the JPA merge, so the guard and update tests assert against rows shaped the way production shapes them.TbRpcServiceTestasserts no rule-engine message on a collision.DeviceActorMessageProcessorTestcovers the duplicate skipping both the send and the pending registration while still replying with therpcId, the expired-on-arrival reply, and a first delivery behaving exactly as before.Not covered: there is no end-to-end test composing the actor branch with the real DAO — the actor tests stub
TbRpcService, andJpaRpcDaoTesthas no actor. A second delivery of the samerpcIdproducing exactly one row and one device publish is verified only as two separately-mocked halves.For the batching, the ordering tests are the point:
DeviceActorMessageProcessorTestregisters two commands and delivers the second one's persist result first, asserts nothing goes out, then delivers the first and asserts it goes out first — for both sequential strategies. Drop the head-of-line guard and those fail. Also covered:FAILEDsends and replies nothing and leaves no pending entry,DUPLICATEreplies therpcIdwithout sending, a failed head doesn't stall the queue, BURST sends only its own command per continuation, and an actor evicted between the two turns still delivers via the reload on next start.JpaRpcDaoTestcovers per-row insert results in a mixed batch, a duplicaterpcIdinside one batch, and an insert plus update for the samerpcIdin one flush applying in order.TbRpcServiceTestcovers the three continuation outcomes and that the actor is resumed before the rule-engine push runs.Also not covered: the load scenario above has not been re-run against this branch yet — that is the acceptance gate for the storage numbers.