diff --git a/CHANGELOG.md b/CHANGELOG.md index e7fc7b0..3fc2dd8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,33 @@ # Changelog +## Unreleased (1.2.0) + +Several application nodes. No wire or contract change (extension 2.4.0, wire v1, subprotocol +unchanged); connectors need no update. Until now the wakeup listeners, the connection +ownership and the claim gate lived in one JVM, so with two nodes a job accepted on the node +without the device's socket was never announced, and a claim that reached that node was +refused with 409. + +- Commit wakeups are handed to the application's QQQ realtime bus (in memory on one node; the + host's shared bus, such as Redis, on several) as well as to this JVM. The realtime route + provider joins the bus at start; hosts add no wiring. The committing request never waits for + the transport. +- A device's one connection is recorded on its row (`realtime_connection_id`, `_node`, `_ready`, + `_expires_at`; four additive columns, added by `initialize()` or by the host's own migration) + with guarded updates. The claim gate on a node without the socket accepts a claim when the + row shows a live, ready connection and the device has applied the current desired state and + settings. +- On every client heartbeat the connection renews its lease and asks the database whether + anything was committed that it has not been told, so a wakeup the shared bus lost is late by + at most one heartbeat interval (20 seconds by default). An idle heartbeat sends no frame. +- A device that connects through another node replaces its previous connection there. +- The QUEUED to CLAIMED update is guarded by the status, in addition to the device row lock. +- The socket upgrade answers 503, not 403, when the database could not answer during + authorization, so a client retries instead of asking a person to sign in. +- Tests: two services with their own buses and route providers over one database and one + application bus, and the claim, the acknowledgement and the connection record on + PostgreSQL 17 (Testcontainers; Docker required to run the suite). + ## 1.1.2 Runtime and publication move to the QRun-IO Gitea Maven registry; no printing contract diff --git a/README.md b/README.md index 9ce57bf..228566b 100644 --- a/README.md +++ b/README.md @@ -92,7 +92,7 @@ After the artifact is available in your configured repository, add: ``` -Supply an existing JDBC DataSource/pool and a QQQ RDBMS backend pointing to that same database. Initial schema is created by `initialize()`; the implementation uses portable H2/PostgreSQL SQL, but only H2 has been tested so far. +Supply an existing JDBC DataSource/pool and a QQQ RDBMS backend pointing to that same database. Initial schema is created by `initialize()`; the implementation uses portable H2/PostgreSQL SQL. H2 carries the full suite; the claim, the acknowledgement, the realtime connection record and `initialize()` itself also run on PostgreSQL 17 in `PrintingClaimPostgresTest` (Docker required). Device and printer names, descriptions and locations are edited in the Windows client and synchronized through heartbeat/inventory. QQQ's native record forms and selectors show the effective friendly name; the physical computer/queue name remains visible in Details. These synchronized fields are read-only in QQQ so two editors do not overwrite each other. Approval remains a separate server control. Friendly names/locations allow 512 characters, descriptions 4096, and physical names/job titles 1024. Clearing a friendly name restores the technical name without changing IDs or routing. @@ -153,7 +153,7 @@ For durable QQQ jobs register `PrintEnqueueEffectTarget` as an `IDEMPOTENT_EXTER ## Deployment boundaries -Jobs target an explicit device/printer. Logical destination routing and a distributed real-time bus belong to the integrating host. The release includes cancellation of queued jobs, explicit reprints, owned spool-cancellation commands, durable receipt reconciliation, S3 delivery storage and scheduled retention. Received/submitting jobs are not automatically offered again; `SPOOLED` never asserts physical delivery. +Jobs target an explicit device/printer. Logical destination routing belongs to the integrating host. An application running as several nodes shares printing through its database and its QQQ realtime bus; see [several application nodes](docs/REALTIME.md#several-application-nodes) (unreleased; 1.1.2 and earlier support one node). The release includes cancellation of queued jobs, explicit reprints, owned spool-cancellation commands, durable receipt reconciliation, S3 delivery storage and scheduled retention. Received/submitting jobs are not automatically offered again; `SPOOLED` never asserts physical delivery. Job paper (`paperWidthMm` and `paperHeightMm`, both positive and at most 2000, never combined with `mediaProfile`) is supported by the matching Windows 1.0.1 connector. The built-in 4 × 1 and 4 × 2 PDF tests use 101.6 × 25.4 / 50.8 mm. Without explicit paper or a reported media selection, jobs follow the Windows queue default. See [media selection](docs/OPERATOR-CATALOG.md). diff --git a/docs/OPERATIONS.md b/docs/OPERATIONS.md index e2b10b9..793454b 100644 --- a/docs/OPERATIONS.md +++ b/docs/OPERATIONS.md @@ -184,4 +184,4 @@ cancellation request, not that paper was withdrawn or a physical print was undon The stable 1.0.0 library and its matching release/registry bytes are published; see [release evidence](RELEASES.md). Local checks cover H2 persistence, native QQQ operations, real HTTP/WebSocket transport, S3-compatible storage and bounded job histories. User-confirmed physical output includes laser PDF and Zebra ZPL/PDF. -A deployment still supplies and validates its own database, S3/IAM, authentication mapping, supervision and printer drivers. PostgreSQL and multiple-node operation are not qualified by the local H2/MinIO fixture. The default real-time bus is in-process; consult [host integration](REALTIME.md) before scaling hosts. These are deployment boundaries, not pending 1.0 publication tasks. +A deployment still supplies and validates its own database, S3/IAM, authentication mapping, supervision and printer drivers. The local H2/MinIO fixture does not qualify a deployment's PostgreSQL or its several nodes. From the release after 1.1.2 the library itself supports several nodes over one database (wakeups on the application's QQQ realtime bus, the device's connection and the claim gate in the database, proven in-process and on PostgreSQL 17); two real nodes behind a load balancer remain the deployment's own qualification. Read [several application nodes](REALTIME.md#several-application-nodes) before scaling hosts. These are deployment boundaries, not pending 1.0 publication tasks. diff --git a/docs/REALTIME.md b/docs/REALTIME.md index 3cc5ec2..c735110 100644 --- a/docs/REALTIME.md +++ b/docs/REALTIME.md @@ -80,20 +80,84 @@ expansion for legacy submission). Raising a document limit above the running transport ceiling requires a host restart or corresponding production ingress configuration. Download streaming does not base64-expand documents. -## Host ownership and deployment boundary +## Several application nodes -Commit wakeups are emitted after SQL commit. A subscription exists before initial -state reconstruction, closing the empty-claim/reconnect notification race. -`PrintingRealtimeBus` is an explicit host-owned notification and connection -ownership seam. The default implementation is in-process; this source does **not** -provide a distributed broker or claim tested multi-node production support. +Unreleased (the next release from `main`); 1.1.2 and earlier support one node only. -Use one writer/application host with its configured service/bus. Multiple route -providers sharing that explicit in-process bus fence a device's prior connection. -A multi-node deployment needs a host-provided shared implementation preserving -both post-commit notifications and single-channel ownership, and its own -qualification. Separate independent service instances do not magically publish -notifications to one another. +An application may run as several nodes (pods) over one database, each with its +own `PrintingService` and its own `PrintingRealtimeRoutes`. A device still holds +one connection, on one node. Nothing about a connection that another node needs +is kept in memory: + +| What | Where it lives | What another node does with it | +|---|---|---| +| Jobs, events, the claim | `printing_job`, `printing_event`; the claim runs under the device row lock and its QUEUED to CLAIMED update is guarded by the status | Claims, cancels and events for one device are serialized across nodes. A job is handed out by exactly one transition; every later claim replays the same assignment until the device records RECEIVED | +| Which connection holds the device | `printing_device.realtime_connection_id`, `realtime_connection_node`, `realtime_connection_ready`, `realtime_connection_expires_at` | The newest connection takes the row. Only the connection named on the row can renew it, mark it ready or clear it, so a late close never removes a replacement. A node that stopped without clearing its row is forgotten when the lease (the heartbeat timeout, 60 seconds by default) runs out | +| Whether a claim may proceed | the same row, plus `control_applied_revision` against `control_revision` and `realtime_applied_revision` against the settings revision | The node holding the socket answers from the connection itself, as before. Any other node accepts the claim when the row shows a live, ready connection and the device has applied the current desired state and settings; otherwise 409, as before | +| Commit wakeups | `PrintingRealtimeBus`: delivered to this JVM directly, and handed to the application's QQQ realtime bus (`RealtimeBus.get(instance)`) on channel `qbit-printing` | Each node's bus receives the wakeup and wakes the connection it holds. The message carries the tenant and device identifiers only. Browsers cannot subscribe to that channel: QQQ realtime streams accept only `instance`, `user`, `table` and `record` channels | + +The route provider joins the application's realtime bus when it starts +(`PrintingRealtimeBus.join(instance)`), so a host adds no wiring. On one node +that bus is QQQ's in-memory bus and behaviour is unchanged. On several nodes +the host configures a shared realtime bus the way it does for the rest of QQQ +(for example qbit-redis `RedisQBitConfig.withRealtime`); without one, each +node's wakeups stay on that node and only the heartbeat reconciliation below +connects them. A service that publishes but has no route provider (a worker +process) joins explicitly with the same call. + +**The shared bus is best effort, and printing does not depend on it.** A +wakeup is handed to the bus by a bounded background worker, so the request that +committed never waits for the transport and a transport failure never fails +it (`getForwardFailures()` counts wakeups that could not be handed over). On +every client heartbeat (`PING`, every `heartbeatIntervalSeconds`, 20 by +default, 5 at the least) the connection renews its lease and asks the database +whether anything was committed that it has not been told: a newer desired +state or settings revision than it last sent, a pending command, or a job a +claim would hand out now. If so it runs the ordinary wakeup. A wakeup the bus +lost is therefore late by at most one heartbeat interval, never lost; an idle +device costs a few indexed reads per heartbeat and receives no frame. When the +transport reports that it may have lost messages (its `resync` hint after a +reconnect or an overflow), every connection a node holds reconciles at once. + +A device that connects through another node replaces its previous connection: +the row changes hands, the other node is told over the bus and closes the old +socket, and if the bus is down the old socket's next frame finds the row named +another connection and closes then. + +Delivery guarantee: a job is delivered to its device at least once (the claim +replays until RECEIVED commits), every event is idempotent by its event id on +any node, and the server never hands out a job that has been received. The +client does not submit a job its journal already holds in a final state, and +reports an interrupted submission as `OUTCOME_UNKNOWN` instead of printing it +again. Together: never two submissions to the spooler for one job. + +The socket upgrade answers 503 when the database could not answer during +authorization, so a client retries; 401 and 403 remain what they were (sign-in +required, denied), because a client stops and asks for a person on those. + +Hosts that own their schema (no `initialize()` at run time) add the four +columns themselves before running this version; they are nullable or +defaulted, so the previous release keeps working on the new schema: + +```sql +ALTER TABLE printing_device ADD COLUMN IF NOT EXISTS realtime_connection_id VARCHAR(36); +ALTER TABLE printing_device ADD COLUMN IF NOT EXISTS realtime_connection_node VARCHAR(200); +ALTER TABLE printing_device ADD COLUMN IF NOT EXISTS realtime_connection_ready INTEGER DEFAULT 0 NOT NULL; +ALTER TABLE printing_device ADD COLUMN IF NOT EXISTS realtime_connection_expires_at BIGINT; +``` + +`realtime_connection_node` (the node id, a pod name in Kubernetes) is shown on +the device record as "Realtime connection node". Lease times are epoch +milliseconds from each node's own clock; nodes are expected to keep ordinary +clock synchronization. + +Verified: `PrintingMultiNodeTest` (two services, buses and route providers over +one database and one application bus: wake, claim over HTTP on the node +without the socket, events on either node, the bus losing or refusing every +message, replacement), `PrintingRealtimeBusTest`, and `PrintingClaimPostgresTest` +(racing claims, concurrent acknowledgements and the connection record on +PostgreSQL 17). Not verified: two real processes behind a load balancer, a +node dying mid-claim, Redis as the transport, and a physical printer. The existing `/printing/v1` routes remain explicit legacy diagnostic compatibility. They are not a real-time fallback and cannot authorize a native bound device. diff --git a/pom.xml b/pom.xml index e50ff50..f9a6387 100644 --- a/pom.xml +++ b/pom.xml @@ -35,6 +35,9 @@ com.h2databaseh22.4.240runtime org.junit.jupiterjunit-jupiter6.0.1test + + org.testcontainerstestcontainers2.0.3test + org.postgresqlpostgresql42.7.8test ui-demo com.kingsrook.qqqqqq-frontend-next1.0.0-RC.38runtime diff --git a/src/main/java/com/kingsrook/qbits/printing/PrintingQBitProducer.java b/src/main/java/com/kingsrook/qbits/printing/PrintingQBitProducer.java index 76e7c13..3880f4b 100644 --- a/src/main/java/com/kingsrook/qbits/printing/PrintingQBitProducer.java +++ b/src/main/java/com/kingsrook/qbits/printing/PrintingQBitProducer.java @@ -105,6 +105,7 @@ public class PrintingQBitProducer implements QBitProducer devices.withField(new QFieldMetaData("control_version", QFieldType.INTEGER).withLabel("Agent control version")); devices.withField(new QFieldMetaData("control_capabilities", QFieldType.STRING).withLabel("Agent control capabilities")); devices.withField(new QFieldMetaData("control_last_poll_at", QFieldType.STRING).withLabel("Last control connection")); + devices.withField(new QFieldMetaData("realtime_connection_node", QFieldType.STRING).withLabel("Realtime connection node")); devices.withField(new QFieldMetaData("revoked_at", QFieldType.STRING).withLabel("Access revoked at")); devices.withVirtualField(new QVirtualFieldMetaData("device_status", QFieldType.STRING).withLabel("Device status").withPossibleValueSourceName("printing_deviceStatus").withIsQuerySelectable(true)); devices.withVirtualField(new QVirtualFieldMetaData("control_status", QFieldType.STRING).withLabel("Control application").withPossibleValueSourceName("printing_controlApplication").withIsQuerySelectable(true)); diff --git a/src/main/java/com/kingsrook/qbits/printing/PrintingRealtimeBus.java b/src/main/java/com/kingsrook/qbits/printing/PrintingRealtimeBus.java index ec4b04a..120ef0b 100644 --- a/src/main/java/com/kingsrook/qbits/printing/PrintingRealtimeBus.java +++ b/src/main/java/com/kingsrook/qbits/printing/PrintingRealtimeBus.java @@ -21,16 +21,51 @@ package com.kingsrook.qbits.printing; +import java.io.Serializable; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.Objects; +import java.util.UUID; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; import java.util.function.BiConsumer; +import com.kingsrook.qqq.backend.core.model.metadata.QInstance; +import com.kingsrook.qqq.backend.core.realtime.RealtimeBus; +import com.kingsrook.qqq.backend.core.realtime.RealtimeBusInterface; +import com.kingsrook.qqq.backend.core.realtime.RealtimeMessage; import org.jspecify.annotations.Nullable; -/** Host-owned commit notification bridge. Callbacks only schedule bounded work. */ +/*************************************************************************** + ** Host-owned commit notification bridge. Callbacks only schedule bounded work. + ** + ** A wakeup always reaches the listeners of this JVM directly. Once joined to + ** the application's QQQ realtime bus (see join), it is also handed to that + ** bus, so the node holding a device's connection is told whichever node + ** committed the change. The shared bus is best effort: a wakeup it loses is + ** recovered by the connection's own reconciliation against the database on + ** each heartbeat, never by this class. A message carries a tenant and a + ** device identifier only. + ***************************************************************************/ public class PrintingRealtimeBus { + /** Not a browser channel: QQQ realtime streams accept only instance, user, table and record channels. */ + static final String SHARED_CHANNEL = "qbit-printing"; + static final String SHARED_EVENT = "printing.changed"; + private static final int MAX_QUEUED_FORWARDS = 1024; private final CopyOnWriteArrayList> listeners = new CopyOnWriteArrayList<>(); - private final java.util.concurrent.ConcurrentHashMap owners = new java.util.concurrent.ConcurrentHashMap<>(); + private final CopyOnWriteArrayList gapListeners = new CopyOnWriteArrayList<>(); + private final ConcurrentHashMap owners = new ConcurrentHashMap<>(); + private final String origin = UUID.randomUUID().toString(); + private final AtomicLong forwardFailures = new AtomicLong(); + private @Nullable Link link; + private int joins; /** Atomically fence the preceding connection across route providers sharing this host bus. */ @@ -65,6 +100,12 @@ public class PrintingRealtimeBus } + /** The shared transport, this bus's subscription to it and the bounded worker that hands wakeups to it. */ + private record Link(RealtimeBusInterface transport, AutoCloseable subscription, ExecutorService forwarder) + { + } + + /** Register before reading the durable snapshot to avoid a reconnect lost wakeup. */ public AutoCloseable subscribe(BiConsumer listener) { @@ -73,8 +114,135 @@ public class PrintingRealtimeBus } + /*************************************************************************** + ** Be told when the shared bus reports that it may have lost messages (its + ** resync hint after a reconnect or an overflow), so every local connection + ** can reconcile against the database at once. + ***************************************************************************/ + public AutoCloseable subscribeGaps(Runnable listener) + { + gapListeners.add(listener); + return () -> gapListeners.remove(listener); + } + + + /*************************************************************************** + ** Join the application's QQQ realtime bus: in memory on one node, the + ** configured shared bus (such as Redis) on several. The realtime route + ** provider does this itself at start; a host calls it only for a service + ** that publishes without a route provider. Joining again shares the one + ** link; close every returned handle to leave. One bus joins one application. + ***************************************************************************/ + public synchronized AutoCloseable join(QInstance instance) + { + if(link == null) + { + RealtimeBusInterface transport = RealtimeBus.get(Objects.requireNonNull(instance)); + ThreadPoolExecutor forwarder = new ThreadPoolExecutor(1, 1, 0, TimeUnit.SECONDS, new ArrayBlockingQueue<>(MAX_QUEUED_FORWARDS), + Thread.ofVirtual().name("printing-realtime-forward").factory(), (task, executor) -> forwardFailures.incrementAndGet()); + link = new Link(transport, transport.subscribe(this::received), forwarder); + } + joins++; + AtomicBoolean left = new AtomicBoolean(); + return () -> + { + if(left.compareAndSet(false, true)) + { + leave(); + } + }; + } + + + /** Drop the link when its last user leaves; queued wakeups are abandoned because the database remains authoritative. */ + private synchronized void leave() throws Exception + { + Link current = link; + if(--joins > 0 || current == null) + { + return; + } + link = null; + current.forwarder().shutdownNow(); + current.subscription().close(); + } + + + /** Wakeups this bus could not hand to the shared bus (full queue or transport error); connections recover them by reconciliation. */ + public long getForwardFailures() + { + return forwardFailures.get(); + } + + /** Publish only after commit. Shared-host implementations must preserve this nonblocking contract. */ public void publish(String tenant, @Nullable String device) + { + deliver(tenant, device); + Link current; + synchronized(this) + { + current = link; + } + if(current != null) + { + // The transport may wait on its network; the committing caller never does. + current.forwarder().execute(() -> forward(current.transport(), tenant, device)); + } + } + + + /** Hand one wakeup to the shared bus, marked with this bus so its own echo is not delivered twice. */ + private void forward(RealtimeBusInterface transport, String tenant, @Nullable String device) + { + try + { + Map data = new LinkedHashMap<>(); + data.put("origin", origin); + data.put("tenant", tenant); + if(device != null) + { + data.put("device", device); + } + transport.publish(new RealtimeMessage().withChannel(SHARED_CHANNEL).withEvent(SHARED_EVENT).withData(data)); + } + catch(RuntimeException exception) + { + forwardFailures.incrementAndGet(); + } + } + + + /** Accept only this QBit's wakeups from other buses, and the transport's own lost-message hint. */ + private void received(RealtimeMessage message) + { + if(message.getAudience() == null && "instance".equals(message.getChannel()) && "resync".equals(message.getEvent())) + { + for(Runnable listener : gapListeners) + { + try + { + listener.run(); + } + catch(RuntimeException ignored) + { + // A stopped listener must not stop the others; each connection also reconciles on its own heartbeat. + } + } + return; + } + Map data = message.getData(); + if(!SHARED_CHANNEL.equals(message.getChannel()) || !SHARED_EVENT.equals(message.getEvent()) || data == null + || !(data.get("origin") instanceof String sender) || origin.equals(sender) || !(data.get("tenant") instanceof String tenant) || tenant.isBlank()) + { + return; + } + deliver(tenant, data.get("device") instanceof String device && !device.isBlank() ? device : null); + } + + + /** Run this JVM's listeners; none of them performs I/O on the calling thread. */ + private void deliver(String tenant, @Nullable String device) { for(BiConsumer listener : listeners) { diff --git a/src/main/java/com/kingsrook/qbits/printing/PrintingRealtimeConnections.java b/src/main/java/com/kingsrook/qbits/printing/PrintingRealtimeConnections.java new file mode 100644 index 0000000..d20088a --- /dev/null +++ b/src/main/java/com/kingsrook/qbits/printing/PrintingRealtimeConnections.java @@ -0,0 +1,128 @@ +/* + * QQQ - Low-code Application Framework for Engineers. + * Copyright (C) 2021-2025. Kingsrook, LLC + * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States + * contact@kingsrook.com + * https://github.com/Kingsrook/ + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.kingsrook.qbits.printing; + + +import javax.sql.DataSource; +import java.sql.Connection; +import java.sql.SQLException; +import org.jspecify.annotations.Nullable; + + +/*************************************************************************** + ** The one realtime connection a device holds, recorded on its row so every + ** node of an application reads the same answer. Each statement is a single + ** guarded update: the newest connection takes the row, and only the + ** connection named on the row can renew, mark or clear it. A node that + ** stops without clearing its row is forgotten when the lease expires. + ** + ** Lease times are epoch milliseconds from the writing node's clock, compared + ** with the reading node's clock; nodes are expected to keep ordinary clock + ** synchronization, and the lease (the heartbeat timeout) is tens of seconds. + ***************************************************************************/ +final class PrintingRealtimeConnections +{ + private final DataSource source; + private final String tenant; + + + /** Bind to the same immutable tenant and pool as the queue. */ + PrintingRealtimeConnections(DataSource source, String tenant) + { + this.source = source; + this.tenant = tenant; + } + + + /** Additive, nullable or defaulted columns: a previous release keeps working on the new schema. */ + static void initialize(Connection connection) throws SQLException + { + for(String sql : new String[] { + "ALTER TABLE printing_device ADD COLUMN IF NOT EXISTS realtime_connection_id VARCHAR(36)", + "ALTER TABLE printing_device ADD COLUMN IF NOT EXISTS realtime_connection_node VARCHAR(200)", + "ALTER TABLE printing_device ADD COLUMN IF NOT EXISTS realtime_connection_ready INTEGER DEFAULT 0 NOT NULL", + "ALTER TABLE printing_device ADD COLUMN IF NOT EXISTS realtime_connection_expires_at BIGINT" + }) + { + PrintingService.update(connection, sql); + } + } + + + /** The newest authenticated connection takes the device; it must apply state and settings again before it is ready. */ + void acquire(String device, String connectionId, String node, long expiresAt) throws SQLException + { + try(Connection connection = source.getConnection()) + { + PrintingService.update(connection, "UPDATE printing_device SET realtime_connection_id=?, realtime_connection_node=?, realtime_connection_ready=0, realtime_connection_expires_at=? WHERE tenant_id=? AND id=?", + connectionId, node.length() > 200 ? node.substring(0, 200) : node, expiresAt, tenant, device); + } + } + + + /** Extend the lease; false means another connection, on this node or another, has taken the device. */ + boolean renew(String device, String connectionId, long expiresAt) throws SQLException + { + try(Connection connection = source.getConnection()) + { + return PrintingService.update(connection, "UPDATE printing_device SET realtime_connection_expires_at=? WHERE tenant_id=? AND id=? AND realtime_connection_id=?", expiresAt, tenant, device, connectionId) == 1; + } + } + + + /** Record that this connection acknowledged everything it was sent; false means it was replaced first. */ + boolean ready(String device, String connectionId, long expiresAt) throws SQLException + { + try(Connection connection = source.getConnection()) + { + return PrintingService.update(connection, "UPDATE printing_device SET realtime_connection_ready=1, realtime_connection_expires_at=? WHERE tenant_id=? AND id=? AND realtime_connection_id=?", expiresAt, tenant, device, connectionId) == 1; + } + } + + + /** Clear only this connection's row; a replacement is never removed by a late close. */ + void release(String device, String connectionId) throws SQLException + { + try(Connection connection = source.getConnection()) + { + PrintingService.update(connection, "UPDATE printing_device SET realtime_connection_id=NULL, realtime_connection_node=NULL, realtime_connection_ready=0, realtime_connection_expires_at=NULL WHERE tenant_id=? AND id=? AND realtime_connection_id=?", + tenant, device, connectionId); + } + } + + + /*************************************************************************** + ** Whether a connection other than the caller's own holds the device, has + ** an unexpired lease, and the device has acknowledged the current desired + ** state and the current settings. This is the claim gate for a request + ** that reached a node which does not hold the device's socket. + ***************************************************************************/ + boolean readyElsewhere(String device, @Nullable String localConnectionId, long now) throws SQLException + { + try(Connection connection = source.getConnection()) + { + return !PrintingService.query(connection, "SELECT d.id FROM printing_device d JOIN printing_settings s ON s.tenant_id=d.tenant_id WHERE d.tenant_id=? AND d.id=?" + + " AND d.enabled=1 AND d.revoked_at IS NULL AND d.removed_at IS NULL AND d.realtime_connection_id IS NOT NULL AND d.realtime_connection_id<>?" + + " AND d.realtime_connection_ready=1 AND d.realtime_connection_expires_at>? AND d.control_applied_revision=d.control_revision AND d.realtime_applied_revision=s.realtime_revision", + tenant, device, localConnectionId == null ? "" : localConnectionId, now).isEmpty(); + } + } +} diff --git a/src/main/java/com/kingsrook/qbits/printing/PrintingRealtimeRoutes.java b/src/main/java/com/kingsrook/qbits/printing/PrintingRealtimeRoutes.java index 60ad0d5..b837122 100644 --- a/src/main/java/com/kingsrook/qbits/printing/PrintingRealtimeRoutes.java +++ b/src/main/java/com/kingsrook/qbits/printing/PrintingRealtimeRoutes.java @@ -36,6 +36,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import com.kingsrook.qqq.backend.core.actions.profile.UserProfiles; +import com.kingsrook.qqq.backend.core.cluster.QNodeRoles; import com.kingsrook.qqq.backend.core.context.QContext; import com.kingsrook.qqq.backend.core.exceptions.QAuthenticationException; import com.kingsrook.qqq.backend.core.exceptions.QPermissionDeniedException; @@ -55,7 +56,18 @@ import org.json.JSONObject; import org.jspecify.annotations.Nullable; -/** Current-user authenticated outbound channels. SQL is authoritative; notifications are only wakeups. */ +/*************************************************************************** + ** Current-user authenticated outbound channels. SQL is authoritative; + ** notifications are only wakeups. + ** + ** Several nodes of one application may each run this provider over the same + ** database. A device holds one connection, on one node, and that connection + ** is recorded on the device row (PrintingRealtimeConnections), so a claim + ** that reaches another node is judged by the same durable state. Commit + ** wakeups travel between nodes on the application's QQQ realtime bus, which + ** this provider joins when it starts; each connection also reconciles with + ** the database on every heartbeat, so a wakeup the bus lost is only late. + ***************************************************************************/ public final class PrintingRealtimeRoutes implements QJavalinRouteProviderInterface, AutoCloseable { public static final String PATH = "/qqq/v1/qbit/printing/device/realtime"; @@ -66,7 +78,10 @@ public final class PrintingRealtimeRoutes implements QJavalinRouteProviderInterf private final Map channels = new ConcurrentHashMap<>(); private final ExecutorService workers = Executors.newVirtualThreadPerTaskExecutor(); private final AutoCloseable subscription; + private final AutoCloseable gapSubscription; + private final String node = QNodeRoles.nodeId(); private @MonotonicNonNull QInstance instance; + private @Nullable AutoCloseable sharedBus; /** Register this provider alongside the host's ordinary native QQQ endpoints. */ @@ -81,6 +96,8 @@ public final class PrintingRealtimeRoutes implements QJavalinRouteProviderInterf channels.values().stream().filter(channel -> device == null || device.equals(channel.device)).forEach(Channel::changed); } }); + // The shared bus says it may have lost messages: every connection held here re-reads the database. + gapSubscription = service.getRealtimeBus().subscribeGaps(() -> channels.values().forEach(Channel::changed)); } @@ -94,6 +111,7 @@ public final class PrintingRealtimeRoutes implements QJavalinRouteProviderInterf @Override public void acceptJavalinConfig(JavalinConfig config) { + joinSharedBus(); config.routes.wsBeforeUpgrade(PATH, context -> { try @@ -109,7 +127,8 @@ public final class PrintingRealtimeRoutes implements QJavalinRouteProviderInterf } catch(Exception exception) { - throw new HttpResponseException(exception instanceof QAuthenticationException ? 401 : 403, "Realtime authentication or origin denied", Map.of()); + int status = upgradeStatus(exception); + throw new HttpResponseException(status, status == 503 ? "Realtime authorization is temporarily unavailable" : "Realtime authentication or origin denied", Map.of()); } finally { @@ -123,12 +142,16 @@ public final class PrintingRealtimeRoutes implements QJavalinRouteProviderInterf String device = Objects.requireNonNull(context.attribute(DEVICE_ATTRIBUTE), "WebSocket connected without the authorized device attribute"); Channel next = new Channel(context, device, context.attribute("printingRealtimeCredential")); next.policy = service.realtimePolicy(); + // The device row names the one connection every node honors. Taking it here fences a connection another node still holds. + service.realtimeConnections().acquire(device, next.generation, node, next.leaseExpiry()); Channel previous = channels.put(device, next); if(previous != null) { previous.stop("Connection replaced"); } service.getRealtimeBus().acquire(service.getTenantId(), device, next.generation, () -> next.stop("Connection replaced")); + // Tell the other nodes now, so one still holding the previous socket of this device closes it without waiting for its next frame. + service.getRealtimeBus().publish(service.getTenantId(), device); context.session.setMaxTextMessageSize(next.policy.get("maxFrameBytes")); context.session.setMaxBinaryMessageSize(0); context.session.setIdleTimeout(Duration.ofMillis(next.policy.get("webSocketHandshakeTimeoutMillis"))); @@ -175,13 +198,53 @@ public final class PrintingRealtimeRoutes implements QJavalinRouteProviderInterf } - /** Gate native claims on acknowledged current settings. */ + /*************************************************************************** + ** Join the application's QQQ realtime bus once the host has supplied its + ** instance, so commit wakeups cross nodes. On one node that bus is in + ** memory and nothing changes. + ***************************************************************************/ + private synchronized void joinSharedBus() + { + if(sharedBus == null && instance != null) + { + sharedBus = service.getRealtimeBus().join(instance); + } + } + /*************************************************************************** + ** Refuse an upgrade as retryable (503) when the database could not answer, + ** as sign-in required (401) when the session is not accepted, and as + ** denied (403) otherwise. A client stops and asks for a person on 401 and + ** 403, so an outage must never be reported as either. + ***************************************************************************/ + static int upgradeStatus(Exception exception) + { + for(Throwable cause = exception; cause != null; cause = cause.getCause()) + { + if(cause instanceof java.sql.SQLException) + { + return 503; + } + } + return exception instanceof QAuthenticationException ? 401 : 403; + } + + + /*************************************************************************** + ** Gate native claims on acknowledged current state and settings. The + ** connection held here answers for itself. A claim that reached a node + ** which does not hold the device's connection is accepted when the device + ** row shows a live, ready connection on another node. + ***************************************************************************/ private void requireReady(String device) throws java.sql.SQLException { Channel channel = channels.get(device); - if(channel == null || !channel.ready(service.getRealtimeSettings().getLong("revision"))) + if(channel != null && channel.ready(service.getRealtimeSettings().getLong("revision"))) + { + return; + } + if(!service.realtimeConnections().readyElsewhere(device, channel == null ? null : channel.generation, System.currentTimeMillis())) { throw new PrintingException(409, "An authenticated realtime connection must apply current state and settings before claiming work"); } @@ -273,6 +336,17 @@ public final class PrintingRealtimeRoutes implements QJavalinRouteProviderInterf try { subscription.close(); + gapSubscription.close(); + AutoCloseable joined; + synchronized(this) + { + joined = sharedBus; + sharedBus = null; + } + if(joined != null) + { + joined.close(); + } } catch(Exception exception) { @@ -364,6 +438,63 @@ public final class PrintingRealtimeRoutes implements QJavalinRouteProviderInterf } + /** The lease this connection writes on its device row: one heartbeat timeout from now, renewed on every heartbeat. */ + private long leaseExpiry() + { + return System.currentTimeMillis() + policy.get("heartbeatTimeoutSeconds") * 1000; + } + + + /** Extend the device row lease. A connection that another node replaced learns it here and stops. */ + private boolean renew() throws java.sql.SQLException + { + if(service.realtimeConnections().renew(device, generation, leaseExpiry())) + { + return true; + } + stop("Connection replaced"); + return false; + } + + + /*************************************************************************** + ** Heartbeat reconciliation. Keep the lease, then ask the database whether + ** anything was committed that this connection has not been told (a commit + ** on another node whose wakeup never arrived). If so, run the ordinary + ** wakeup; otherwise only printer approval is compared. An idle device + ** costs a few indexed reads per heartbeat and receives no frame. + ***************************************************************************/ + private void reconcile() throws Exception + { + if(!renew()) + { + return; + } + long state; + long settings; + synchronized(this) + { + state = sentState; + settings = sentSettings; + } + if(service.realtimePending(device, state, settings)) + { + changed(); + } + else + { + printers(true, false); + } + } + + + /** Everything sent on this connection has been acknowledged by the device. */ + private synchronized boolean caughtUp() + { + return hello != null && appliedState == sentState && appliedSettings == sentSettings; + } + + /** Read synchronized connection application state. */ @@ -448,6 +579,10 @@ public final class PrintingRealtimeRoutes implements QJavalinRouteProviderInterf authorizeCurrent(); if(frame == null) { + if(!renew()) + { + return; + } snapshot(false, helloPayload()); } else @@ -514,6 +649,12 @@ public final class PrintingRealtimeRoutes implements QJavalinRouteProviderInterf appliedState = state; appliedSettings = settings; } + // Durable readiness lets a claim that reaches another node pass its gate; only the connection named on the row may set it. + if(caughtUp() && !service.realtimeConnections().ready(device, generation, leaseExpiry())) + { + stop("Connection replaced"); + return; + } changed(); } case "JOB_EVENT" -> @@ -543,6 +684,7 @@ public final class PrintingRealtimeRoutes implements QJavalinRouteProviderInterf { service.realtimeContact(device); send("PONG", new JSONObject().put("serverTime", Instant.now().toString())); + reconcile(); return; } default -> throw new PrintingException(400, "Unsupported realtime message"); @@ -745,10 +887,14 @@ public final class PrintingRealtimeRoutes implements QJavalinRouteProviderInterf /** Fence and remove this connection without affecting a replacement. */ - private synchronized void stop(String reason) + private void stop(String reason) { - if(!stopped) + synchronized(this) { + if(stopped) + { + return; + } stopped = true; mailbox.clear(); queuedBytes = 0; @@ -756,6 +902,15 @@ public final class PrintingRealtimeRoutes implements QJavalinRouteProviderInterf service.getRealtimeBus().release(service.getTenantId(), device, generation); socket.closeSession(1008, reason); } + try + { + service.realtimeConnections().release(device, generation); + } + catch(java.sql.SQLException | RuntimeException exception) + { + // The row is cleared only by its own connection; one that could not clear it is forgotten when its lease expires. + System.getLogger(PrintingRealtimeRoutes.class.getName()).log(System.Logger.Level.WARNING, "Realtime connection record was not cleared: " + exception.getClass().getSimpleName()); + } } } diff --git a/src/main/java/com/kingsrook/qbits/printing/PrintingService.java b/src/main/java/com/kingsrook/qbits/printing/PrintingService.java index dcf0f80..4f6942a 100644 --- a/src/main/java/com/kingsrook/qbits/printing/PrintingService.java +++ b/src/main/java/com/kingsrook/qbits/printing/PrintingService.java @@ -119,6 +119,47 @@ public class PrintingService } + /** The durable record of each device's one realtime connection, shared by every node using this database. */ + PrintingRealtimeConnections realtimeConnections() + { + return new PrintingRealtimeConnections(dataSource, tenantId); + } + + + /*************************************************************************** + ** Whether the database holds something a connection has not been told: + ** a newer desired state or settings revision than it last sent, a pending + ** command, or a job a claim would hand out now. A connection asks on each + ** heartbeat, so a commit wakeup that never arrived (it was published on + ** another node while the shared bus was unavailable) costs at most one + ** heartbeat interval. Read only; it takes no row lock. + ***************************************************************************/ + boolean realtimePending(String deviceId, long sentState, long sentSettings) throws SQLException + { + try(Connection connection = dataSource.getConnection()) + { + List rows = query(connection, "SELECT d.dispatch_paused, d.connection_state, d.control_capabilities, d.control_revision, s.realtime_revision FROM printing_device d JOIN printing_settings s ON s.tenant_id=d.tenant_id" + + " WHERE d.tenant_id=? AND d.id=? AND d.enabled=1 AND d.revoked_at IS NULL AND d.removed_at IS NULL", tenantId, deviceId); + if(rows.isEmpty()) + { + return false; + } + JSONObject control = rows.getFirst(); + if(control.getLong("control_revision") != sentState || control.getLong("realtime_revision") != sentSettings) + { + return true; + } + if(!query(connection, "SELECT id FROM printing_device_command WHERE tenant_id=? AND device_id=? AND status='PENDING' AND expires_at>=? AND (action<>'REFRESH_INVENTORY' OR ?='CONNECTED') LIMIT 1", + tenantId, deviceId, Instant.now().toString(), control.getString("connection_state")).isEmpty()) + { + return true; + } + JSONObject candidate = dispatchCandidate(connection, deviceId, null, control); + return candidate != null && mediaSupported(candidate, control); + } + } + + /** A host may share this event bus across services using the same database. */ public PrintingService withRealtimeBus(PrintingRealtimeBus bus) { @@ -340,6 +381,7 @@ public class PrintingService } PrintingAdmission.initialize(connection, tenantId); PrintingDeviceControls.initialize(connection); + PrintingRealtimeConnections.initialize(connection); update(connection, "ALTER TABLE printing_device ADD COLUMN IF NOT EXISTS bound_user_id VARCHAR(200)"); widenColumn(connection, "printing_device", "name", 1024); widenColumn(connection, "printing_enrollment", "device_name", 1024); @@ -1466,6 +1508,26 @@ public class PrintingService /** Native adapters authorize one exact job and printer before this locked dispatch transition. */ public JSONObject claimForJob(String deviceId, @Nullable String authorizedJobId) throws SQLException + { + return claimAttempt(deviceId, authorizedJobId).response(); + } + + + /** One claim's wire response, and whether this call moved the job from QUEUED to CLAIMED (false for an empty queue or a replayed assignment). */ + record ClaimAttempt(JSONObject response, boolean transitioned) + { + } + + + /*************************************************************************** + ** Claim under the device row lock, which serializes every claim, cancel + ** and event for the device across all nodes sharing the database. The + ** status guard on the update makes the QUEUED to CLAIMED transition + ** happen once even for a writer that did not take that lock: a job is + ** handed out by exactly one transition, and every later claim replays the + ** same assignment until the device records RECEIVED. + ***************************************************************************/ + ClaimAttempt claimAttempt(String deviceId, @Nullable String authorizedJobId) throws SQLException { try(Connection connection = dataSource.getConnection()) { @@ -1476,30 +1538,28 @@ public class PrintingService query(connection, "SELECT id FROM printing_device WHERE id=? FOR UPDATE", deviceId); requireDevice(connection, deviceId); JSONObject control = query(connection, "SELECT dispatch_paused,connection_state,control_capabilities FROM printing_device WHERE tenant_id=? AND id=?", tenantId, deviceId).getFirst(); - if(control.getInt("dispatch_paused") == 1 || !"CONNECTED".equals(control.getString("connection_state"))) + JSONObject selected = dispatchCandidate(connection, deviceId, authorizedJobId, control); + if(selected == null) { connection.commit(); - return new JSONObject().put("job", JSONObject.NULL); + return new ClaimAttempt(new JSONObject().put("job", JSONObject.NULL), false); } - List candidates = query(connection, "SELECT j." + JOB_COLUMNS.replace(", ", ", j.") + " FROM printing_job j JOIN printing_printer p ON p.device_id=j.device_id AND p.printer_id=j.printer_id WHERE j.tenant_id=? AND j.device_id=? AND p.enabled=1 AND p.available=1 AND p.removed_at IS NULL AND j.status IN ('QUEUED','CLAIMED') AND ((j.format='PDF' AND p.formats LIKE '%\"PDF\"%') OR (j.format='ZPL' AND p.formats LIKE '%\"ZPL\"%'))" + (authorizedJobId == null ? "" : " AND j.id=?") + " ORDER BY j.created_at, j.id LIMIT 1", authorizedJobId == null ? new Object[] { tenantId, deviceId } : new Object[] { tenantId, deviceId, authorizedJobId }); - if(candidates.isEmpty()) - { - connection.commit(); - return new JSONObject().put("job", JSONObject.NULL); - } - JSONObject selected = candidates.getFirst(); - if(new JSONObject(selected.getString("options_json")).has("mediaProfile") - && !new JSONArray(control.optString("control_capabilities", "[]")).toList().contains("printer-media-v1")) + if(!mediaSupported(selected, control)) { throw new PrintingException(409, "This job requires an agent that validates printer media profiles"); } + boolean transitioned = false; if("QUEUED".equals(selected.getString("status"))) { - update(connection, "UPDATE printing_job SET status='CLAIMED' WHERE id=?", selected.getString("id")); + if(update(connection, "UPDATE printing_job SET status='CLAIMED' WHERE id=? AND tenant_id=? AND status='QUEUED'", selected.getString("id"), tenantId) != 1) + { + throw new PrintingException(409, "The job changed before it could be claimed"); + } selected.put("status", "CLAIMED"); + transitioned = true; } connection.commit(); - return new JSONObject().put("job", job(selected)); + return new ClaimAttempt(new JSONObject().put("job", job(selected)), transitioned); } catch(Exception exception) { @@ -1510,6 +1570,32 @@ public class PrintingService } + /*************************************************************************** + ** The one job a claim would hand this device now: the oldest QUEUED or + ** CLAIMED job for an approved, exposed printer that accepts its format. + ** Null while dispatch is paused, the device is asked to disconnect or + ** nothing qualifies. The claim and the heartbeat reconciliation share it, + ** so a connection is only told about work a claim would return. + ***************************************************************************/ + private @Nullable JSONObject dispatchCandidate(Connection connection, String deviceId, @Nullable String authorizedJobId, JSONObject control) throws SQLException + { + if(control.getInt("dispatch_paused") == 1 || !"CONNECTED".equals(control.getString("connection_state"))) + { + return null; + } + List candidates = query(connection, "SELECT j." + JOB_COLUMNS.replace(", ", ", j.") + " FROM printing_job j JOIN printing_printer p ON p.device_id=j.device_id AND p.printer_id=j.printer_id WHERE j.tenant_id=? AND j.device_id=? AND p.enabled=1 AND p.available=1 AND p.removed_at IS NULL AND j.status IN ('QUEUED','CLAIMED') AND ((j.format='PDF' AND p.formats LIKE '%\"PDF\"%') OR (j.format='ZPL' AND p.formats LIKE '%\"ZPL\"%'))" + (authorizedJobId == null ? "" : " AND j.id=?") + " ORDER BY j.created_at, j.id LIMIT 1", authorizedJobId == null ? new Object[] { tenantId, deviceId } : new Object[] { tenantId, deviceId, authorizedJobId }); + return candidates.isEmpty() ? null : candidates.getFirst(); + } + + + /** A job that names a media profile needs an agent that validates it. */ + private static boolean mediaSupported(JSONObject job, JSONObject control) + { + return !new JSONObject(job.getString("options_json")).has("mediaProfile") + || new JSONArray(control.optString("control_capabilities", "[]")).toList().contains("printer-media-v1"); + } + + /*************************************************************************** ** Fetch content only for this device's assigned job. diff --git a/src/test/java/com/kingsrook/qbits/printing/PrintingClaimPostgresTest.java b/src/test/java/com/kingsrook/qbits/printing/PrintingClaimPostgresTest.java new file mode 100644 index 0000000..8b958c6 --- /dev/null +++ b/src/test/java/com/kingsrook/qbits/printing/PrintingClaimPostgresTest.java @@ -0,0 +1,337 @@ +/* + * QQQ - Low-code Application Framework for Engineers. + * Copyright (C) 2021-2025. Kingsrook, LLC + * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States + * contact@kingsrook.com + * https://github.com/Kingsrook/ + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.kingsrook.qbits.printing; + + +import java.nio.charset.StandardCharsets; +import java.sql.Connection; +import java.sql.SQLException; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Base64; +import java.util.List; +import java.util.UUID; +import java.util.concurrent.Callable; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import org.json.JSONArray; +import org.json.JSONObject; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.wait.strategy.Wait; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + + +/*************************************************************************** + ** The claim, the acknowledgement and the connection record on real + ** PostgreSQL, through two PrintingService instances with separate + ** connections, as two nodes of one application use them. The row lock and + ** the guarded updates are what make these safe across nodes, so they are + ** proven on the production database and not on H2. Requires Docker. + ***************************************************************************/ +class PrintingClaimPostgresTest +{ + private static final String PASSWORD = "printing-fixture-only"; + private static GenericContainer postgres; + private PrintingService first; + private PrintingService second; + private DriverDataSource source; + private String device; + + + /** An isolated database container is required; never use an application database. */ + @BeforeAll + static void startPostgres() + { + postgres = new GenericContainer<>("postgres:17-alpine").withEnv("POSTGRES_PASSWORD", PASSWORD).withExposedPorts(5432) + .waitingFor(Wait.forLogMessage(".*database system is ready to accept connections.*", 2)); + postgres.start(); + } + + + /** Stop only the database owned by this fixture. */ + @AfterAll + static void stopPostgres() + { + if(postgres != null) + { + postgres.stop(); + } + } + + + /** A fresh schema, two services over it, one enrolled device with an approved queue. */ + @BeforeEach + void setUp() throws Exception + { + source = new DriverDataSource("jdbc:postgresql://" + postgres.getHost() + ":" + postgres.getMappedPort(5432) + "/postgres", "postgres", PASSWORD); + try(Connection connection = source.getConnection(); var sql = connection.createStatement()) + { + sql.execute("DROP SCHEMA public CASCADE; CREATE SCHEMA public"); + } + String tenant = "tenant-" + UUID.randomUUID(); + first = new PrintingService(source, tenant); + first.initialize(); + second = new PrintingService(new DriverDataSource(source.url(), source.user(), source.password()), tenant); + JSONObject enrolled = first.enrollForUser(first.createEnrollment("default", "PostgreSQL fixture").put("deviceName", "PostgreSQL fixture").put("agentVersion", "test"), "operator"); + device = enrolled.getString("deviceId"); + first.inventory(device, new JSONArray().put(new JSONObject().put("printerId", "queue").put("name", "Synthetic queue").put("status", "READY").put("formats", new JSONArray().put("ZPL")))); + first.enablePrinter(device, "queue", true); + } + + + /** Many claims racing through both instances hand out one job, by one transition, and never the next job before the first is received. */ + @Test + void racingClaimsThroughTwoInstancesTransitionEachJobOnce() throws Exception + { + try(ExecutorService pool = Executors.newFixedThreadPool(8)) + { + for(int round = 0; round < 12; round++) + { + String job = first.submit(request("race-" + round)).getString("jobId"); + String next = second.submit(request("race-next-" + round)).getString("jobId"); + CountDownLatch start = new CountDownLatch(1); + List> attempts = new ArrayList<>(); + for(int index = 0; index < 8; index++) + { + PrintingService service = index % 2 == 0 ? first : second; + Callable claim = () -> + { + start.await(); + return service.claimAttempt(device, null); + }; + attempts.add(pool.submit(claim)); + } + start.countDown(); + int transitions = 0; + for(Future attempt : attempts) + { + PrintingService.ClaimAttempt result = attempt.get(30, TimeUnit.SECONDS); + assertEquals(job, result.response().getJSONObject("job").getString("jobId"), "Every racing claim is answered with the same, oldest job"); + assertEquals("CLAIMED", result.response().getJSONObject("job").getString("status")); + transitions += result.transitioned() ? 1 : 0; + } + assertEquals(1, transitions, "Exactly one claim moved the job from QUEUED to CLAIMED"); + assertEquals("QUEUED", first.getJob(next).getString("status"), "The next job is not handed out while the first is unacknowledged"); + + // The acknowledgement arrives through the other instance; the job is then never handed out again by either. + second.event(device, job, new JSONObject().put("eventId", "received-" + round).put("status", "RECEIVED")); + PrintingService.ClaimAttempt following = first.claimAttempt(device, null); + assertEquals(next, following.response().getJSONObject("job").getString("jobId")); + assertTrue(following.transitioned()); + first.event(device, next, new JSONObject().put("eventId", "received-next-" + round).put("status", "RECEIVED")); + assertTrue(second.claim(device).isNull("job")); + assertFalse(first.claimAttempt(device, null).transitioned()); + } + } + } + + + /** The same event delivered through both instances, even at the same instant, is recorded once; a different event under its id is refused. */ + @Test + void acknowledgementThroughEitherInstanceIsRecordedOnce() throws Exception + { + String job = first.submit(request("acknowledge")).getString("jobId"); + assertEquals(job, first.claim(device).getJSONObject("job").getString("jobId")); + JSONObject received = new JSONObject().put("eventId", "ack-received").put("status", "RECEIVED"); + try(ExecutorService pool = Executors.newFixedThreadPool(6)) + { + CountDownLatch start = new CountDownLatch(1); + List> deliveries = new ArrayList<>(); + for(int index = 0; index < 6; index++) + { + PrintingService service = index % 2 == 0 ? first : second; + Callable delivery = () -> + { + start.await(); + service.event(device, job, new JSONObject(received.toString())); + return true; + }; + deliveries.add(pool.submit(delivery)); + } + start.countDown(); + for(Future delivery : deliveries) + { + assertTrue(delivery.get(30, TimeUnit.SECONDS), "A replayed acknowledgement is accepted, never an error"); + } + } + assertEquals(1, count("SELECT COUNT(*) FROM printing_event WHERE job_id=?", job)); + assertEquals("RECEIVED", second.getJob(job).getString("status")); + assertEquals(409, assertThrows(PrintingException.class, () -> second.event(device, job, new JSONObject().put("eventId", "ack-received").put("status", "SPOOLED"))).getStatus()); + second.event(device, job, new JSONObject().put("eventId", "ack-submitting").put("status", "SUBMITTING")); + first.event(device, job, new JSONObject().put("eventId", "ack-spooled").put("status", "SPOOLED").put("spoolerJobId", "17")); + second.event(device, job, new JSONObject().put("eventId", "ack-spooled").put("status", "SPOOLED").put("spoolerJobId", "17")); + assertEquals(409, assertThrows(PrintingException.class, () -> first.event(device, job, new JSONObject().put("eventId", "ack-late").put("status", "SUBMITTING"))).getStatus(), + "A job cannot move backwards through another instance"); + assertEquals("SPOOLED", first.getJob(job).getString("status")); + assertEquals(3, count("SELECT COUNT(*) FROM printing_event WHERE job_id=?", job)); + assertTrue(first.claim(device).isNull("job")); + } + + + /** The connection record: the newest connection takes the device, only the named connection can renew, mark or clear it, and readiness needs all of its parts. */ + @Test + void connectionRecordIsGuardedAndReadinessNeedsEveryPart() throws Exception + { + PrintingRealtimeConnections onFirst = first.realtimeConnections(); + PrintingRealtimeConnections onSecond = second.realtimeConnections(); + long now = System.currentTimeMillis(); + assertFalse(onSecond.readyElsewhere(device, null, now), "No connection"); + onFirst.acquire(device, "connection-1", "node-a", now + 60000); + assertFalse(onSecond.readyElsewhere(device, null, now), "Connected, nothing applied"); + first.applyRealtimeState(device, control(0), first.getRealtimeSettings().getLong("revision")); + assertTrue(onFirst.ready(device, "connection-1", now + 60000)); + assertTrue(onSecond.readyElsewhere(device, null, now)); + assertFalse(onFirst.readyElsewhere(device, "connection-1", now), "A node does not take its own connection for one elsewhere"); + assertFalse(onSecond.readyElsewhere(device, null, now + 60001), "The lease ran out"); + assertEquals("node-a", row().getString("realtime_connection_node")); + + // Another node takes the device: the first connection can no longer renew, mark or clear the row. + onSecond.acquire(device, "connection-2", "node-b", now + 60000); + assertFalse(onFirst.renew(device, "connection-1", now + 120000)); + assertFalse(onFirst.ready(device, "connection-1", now + 120000)); + onFirst.release(device, "connection-1"); + assertEquals("connection-2", row().getString("realtime_connection_id")); + assertEquals(0, row().getInt("realtime_connection_ready"), "A new connection must apply state again"); + assertFalse(onFirst.readyElsewhere(device, null, now)); + assertTrue(onSecond.ready(device, "connection-2", now + 60000)); + assertTrue(onFirst.readyElsewhere(device, "connection-1", now)); + assertTrue(onSecond.renew(device, "connection-2", now + 90000)); + assertEquals(now + 90000, row().getLong("realtime_connection_expires_at")); + + // Newer settings, or a newer desired state, close the gate on every node until the device applies them. + JSONObject policy = first.getRealtimeSettings(); + policy.remove("revision"); + first.saveRealtimeSettings(policy.put("maxCopies", 96)); + assertFalse(onFirst.readyElsewhere(device, null, now)); + second.applyRealtimeState(device, control(0), 1); + assertTrue(onFirst.readyElsewhere(device, null, now)); + first.deviceAction(device, action("PAUSE_DISPATCH"), "operator"); + assertFalse(onFirst.readyElsewhere(device, null, now)); + second.applyRealtimeState(device, control(desiredRevision()), 1); + assertTrue(onFirst.readyElsewhere(device, null, now)); + + onSecond.release(device, "connection-2"); + assertTrue(row().isNull("realtime_connection_id")); + assertTrue(row().isNull("realtime_connection_node")); + assertTrue(row().isNull("realtime_connection_expires_at")); + assertFalse(onFirst.readyElsewhere(device, null, now)); + } + + + /** The heartbeat reconciliation announces exactly what a claim would hand out, and newer state, settings or commands. */ + @Test + void pendingWorkMatchesWhatAClaimWouldReturn() throws Exception + { + long settings = first.getRealtimeSettings().getLong("revision"); + first.deviceControlPoll(device, control(0)); + assertFalse(second.realtimePending(device, 0, settings), "Idle"); + String job = first.submit(request("pending")).getString("jobId"); + assertTrue(second.realtimePending(device, 0, settings), "A queued job for an approved, exposed printer"); + assertEquals(job, second.claim(device).getJSONObject("job").getString("jobId")); + assertTrue(first.realtimePending(device, 0, settings), "A claimed job is replayed until it is received"); + first.event(device, job, new JSONObject().put("eventId", "pending-received").put("status", "RECEIVED")); + assertFalse(second.realtimePending(device, 0, settings), "A received job is not announced"); + + String waiting = first.submit(request("pending-paused")).getString("jobId"); + first.deviceAction(device, action("PAUSE_DISPATCH"), "operator"); + long paused = desiredRevision(); + assertTrue(second.realtimePending(device, 0, settings), "A newer desired state than the connection last sent"); + assertFalse(second.realtimePending(device, paused, settings), "Paused: a claim would return nothing, so nothing is announced"); + first.deviceAction(device, action("RESUME_DISPATCH"), "operator"); + long resumed = desiredRevision(); + assertTrue(second.realtimePending(device, resumed, settings)); + first.enablePrinter(device, "queue", false); + assertFalse(second.realtimePending(device, resumed, settings), "A job for a printer that is no longer approved is not claimable"); + first.enablePrinter(device, "queue", true); + + JSONObject policy = first.getRealtimeSettings(); + policy.remove("revision"); + first.saveRealtimeSettings(policy.put("maxCopies", 95)); + first.cancel(device, waiting); + assertTrue(second.realtimePending(device, resumed, settings), "Newer settings than the connection last sent"); + assertFalse(second.realtimePending(device, resumed, settings + 1)); + first.deviceAction(device, action("REFRESH_INVENTORY"), "operator"); + assertTrue(second.realtimePending(device, desiredRevision(), settings + 1), "A pending command"); + assertFalse(second.realtimePending(UUID.randomUUID().toString(), 0, 0), "An unknown device has nothing pending"); + } + + + /** Read the device row both instances consult. */ + private JSONObject row() throws SQLException + { + try(Connection connection = source.getConnection()) + { + return PrintingService.query(connection, "SELECT realtime_connection_id, realtime_connection_node, realtime_connection_ready, realtime_connection_expires_at FROM printing_device WHERE id=?", device).getFirst(); + } + } + + + /** The desired-state revision an administrator's action left on the device. */ + private long desiredRevision() throws SQLException + { + try(Connection connection = source.getConnection()) + { + return PrintingService.query(connection, "SELECT control_revision FROM printing_device WHERE id=?", device).getFirst().getLong("control_revision"); + } + } + + + /** Count rows for an assertion. */ + private long count(String sql, Object... values) throws SQLException + { + try(Connection connection = source.getConnection()) + { + return ((Number) PrintingService.query(connection, sql, values).getFirst().toMap().values().iterator().next()).longValue(); + } + } + + + /** The control acknowledgement a connected device sends with its applied state. */ + private static JSONObject control(long applied) + { + return new JSONObject().put("controlVersion", 1).put("agentVersion", "test").put("capabilities", new JSONArray().put("REFRESH_INVENTORY").put("CANCEL_SPOOL")).put("lastAppliedRevision", applied); + } + + + /** Stable synthetic submission identity; no physical printer exists in this fixture. */ + private JSONObject request(String key) + { + return new JSONObject().put("deviceId", device).put("printerId", "queue").put("title", "Synthetic PostgreSQL") + .put("format", "ZPL").put("contentBase64", Base64.getEncoder().encodeToString("^XA^XZ".getBytes(StandardCharsets.US_ASCII))).put("idempotencyKey", key); + } + + + /** Native actions retain their existing temporal replay boundary. */ + private static JSONObject action(String name) + { + return new JSONObject().put("action", name).put("requestId", UUID.randomUUID().toString()).put("issuedAt", Instant.now().toString()); + } +} diff --git a/src/test/java/com/kingsrook/qbits/printing/PrintingMultiNodeTest.java b/src/test/java/com/kingsrook/qbits/printing/PrintingMultiNodeTest.java new file mode 100644 index 0000000..42df3df --- /dev/null +++ b/src/test/java/com/kingsrook/qbits/printing/PrintingMultiNodeTest.java @@ -0,0 +1,565 @@ +/* + * QQQ - Low-code Application Framework for Engineers. + * Copyright (C) 2021-2025. Kingsrook, LLC + * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States + * contact@kingsrook.com + * https://github.com/Kingsrook/ + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.kingsrook.qbits.printing; + + +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.nio.charset.StandardCharsets; +import java.nio.file.Path; +import java.sql.Connection; +import java.sql.SQLException; +import java.util.ArrayList; +import java.util.Base64; +import java.util.List; +import java.util.Objects; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Consumer; +import com.kingsrook.qqq.backend.core.exceptions.QAuthenticationException; +import com.kingsrook.qqq.backend.core.exceptions.QException; +import com.kingsrook.qqq.backend.core.exceptions.QPermissionDeniedException; +import com.kingsrook.qqq.backend.core.model.metadata.QInstance; +import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference; +import com.kingsrook.qqq.backend.core.realtime.RealtimeBusInterface; +import com.kingsrook.qqq.backend.core.realtime.RealtimeMessage; +import com.kingsrook.qqq.backend.core.realtime.RealtimeMetaData; +import com.kingsrook.qqq.middleware.javalin.QApplicationLauncher; +import com.kingsrook.qqq.middleware.javalin.QApplicationLauncherConfig; +import io.javalin.Javalin; +import org.json.JSONArray; +import org.json.JSONObject; +import org.jspecify.annotations.Nullable; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + + +/*************************************************************************** + ** Two nodes of one application in one JVM: each has its own PrintingService, + ** its own notification bus and its own realtime route provider (so its own + ** sockets), and they share one database and one QQQ realtime bus, as two + ** pods do. Node A is the full QQQ server with the native HTTP routes; node B + ** serves only the realtime socket. Sessions are real; jobs are synthetic and + ** no printer exists. + ***************************************************************************/ +class PrintingMultiNodeTest +{ + private static final String PREFIX = "/qqq/v1" + PrintingUiApi.PREFIX; + @TempDir + Path directory; + private @Nullable Nodes nodes; + + + /** Every test starts with a working shared bus. */ + @BeforeEach + void resetBus() + { + SwitchableBus.reset(); + } + + + /** Stop both nodes even when an assertion failed. */ + @AfterEach + void stop() + { + if(nodes != null) + { + nodes.close(); + } + } + + + /** A job accepted on the node that does not hold the device's socket wakes the device, and its claim and events are accepted on either node. */ + @Test + void jobAcceptedOnOneNodeWakesTheDeviceAndIsClaimedThroughTheOther() throws Exception + { + Nodes cluster = start("wake"); + try(HttpClient http = HttpClient.newHttpClient()) + { + String cookie = login(http, cluster.baseA); + assertEquals(409, send(http, cluster.baseA, "/device/jobs/claim", "POST", new JSONObject(), cookie, cluster).statusCode(), "No node holds a ready connection yet"); + try(PrintingSocketFixture socket = PrintingSocketFixture.connect(http, cluster.baseB, cookie, cluster.device, cluster.proof)) + { + socket.take("JOB_AVAILABLE"); + socket.received.clear(); + JSONObject row = cluster.deviceRow(); + assertEquals(1, row.getInt("realtime_connection_ready")); + assertFalse(row.isNull("realtime_connection_id")); + assertFalse(row.getString("realtime_connection_node").isBlank()); + + // Accepted by node A, which holds no socket for this device: the wakeup crosses the shared bus to node B. + String job = cluster.serviceA.submit(request(cluster.device, "accepted-on-a")).getString("jobId"); + socket.take("JOB_AVAILABLE"); + assertTrue(SwitchableBus.PUBLISHED.get() > 0, "The wakeup was handed to the application's realtime bus"); + + // The claim reaches node A over HTTP. Its gate passes on the durable connection record, not on a socket of its own. + JSONObject claimed = json(send(http, cluster.baseA, "/device/jobs/claim", "POST", new JSONObject(), cookie, cluster), 200).getJSONObject("job"); + assertEquals(job, claimed.getString("jobId")); + assertEquals("CLAIMED", cluster.serviceB.getJob(job).getString("status")); + assertEquals(job, json(send(http, cluster.baseA, "/device/jobs/claim", "POST", new JSONObject(), cookie, cluster), 200).getJSONObject("job").getString("jobId"), + "An unacknowledged claim replays the same assignment on any node"); + + // Acknowledged over HTTP on node A, then the identical event again over the socket on node B: one event, no error. + JSONObject received = new JSONObject().put("eventId", "multi-node-received").put("status", "RECEIVED"); + json(send(http, cluster.baseA, "/device/jobs/" + job + "/events", "POST", received, cookie, cluster), 200); + socket.send("JOB_EVENT", "event-multi-node-received", new JSONObject().put("jobId", job).put("event", received)); + assertEquals("event-multi-node-received", socket.take("ACK").getJSONObject("payload").getString("acknowledgedMessageId")); + assertEquals("RECEIVED", cluster.serviceA.getJob(job).getString("status")); + assertEquals(1, cluster.count("SELECT COUNT(*) FROM printing_event WHERE job_id=?", job)); + assertTrue(json(send(http, cluster.baseA, "/device/jobs/claim", "POST", new JSONObject(), cookie, cluster), 200).isNull("job"), "A received job is never handed out again"); + + socket.send("JOB_EVENT", "event-multi-node-spooled", new JSONObject().put("jobId", job).put("event", new JSONObject().put("eventId", "multi-node-spooled").put("status", "SPOOLED"))); + socket.take("ACK"); + json(send(http, cluster.baseA, "/device/jobs/" + job + "/events", "POST", new JSONObject().put("eventId", "multi-node-spooled").put("status", "SPOOLED"), cookie, cluster), 200); + assertEquals("SPOOLED", cluster.serviceA.getJob(job).getString("status")); + assertEquals(2, cluster.count("SELECT COUNT(*) FROM printing_event WHERE job_id=?", job)); + } + awaitReleased(cluster); + assertEquals(409, send(http, cluster.baseA, "/device/jobs/claim", "POST", new JSONObject(), cookie, cluster).statusCode(), "A closed connection is not ready on any node"); + } + } + + + /** The gate on a node without the socket follows the device row: handshake, applied revisions and the lease. */ + @Test + void claimGateOnAnotherNodeFollowsTheDurableConnectionRecord() throws Exception + { + Nodes cluster = start("gate"); + try(HttpClient http = HttpClient.newHttpClient()) + { + String cookie = login(http, cluster.baseA); + try(PrintingSocketFixture socket = PrintingSocketFixture.open(http, cluster.baseB, cookie, cluster.device, cluster.proof)) + { + await(() -> !cluster.deviceRow().isNull("realtime_connection_id")); + assertEquals(409, assertThrows(PrintingException.class, () -> cluster.serviceA.requireRealtimeReady(cluster.device)).getStatus(), "Connected but nothing applied"); + socket.initialize(); + socket.take("JOB_AVAILABLE"); + cluster.serviceA.requireRealtimeReady(cluster.device); + cluster.serviceB.requireRealtimeReady(cluster.device); + + // An administrator on node A changes the settings: no node accepts a claim until the device applies them. + JSONObject policy = cluster.serviceA.getRealtimeSettings(); + policy.remove("revision"); + cluster.serviceA.saveRealtimeSettings(policy.put("maxCopies", 98)); + assertEquals(409, assertThrows(PrintingException.class, () -> cluster.serviceA.requireRealtimeReady(cluster.device)).getStatus()); + assertEquals(1, socket.take("SETTINGS").getJSONObject("payload").getLong("revision")); + assertEquals(409, assertThrows(PrintingException.class, () -> cluster.serviceB.requireRealtimeReady(cluster.device)).getStatus()); + socket.apply(0, 1); + cluster.serviceA.requireRealtimeReady(cluster.device); + cluster.serviceB.requireRealtimeReady(cluster.device); + + // Dispatch paused on node A: the desired state reaches the device on node B, and the gate waits for it to be applied. + cluster.serviceA.deviceAction(cluster.device, action("PAUSE_DISPATCH"), "operator"); + assertEquals(409, assertThrows(PrintingException.class, () -> cluster.serviceA.requireRealtimeReady(cluster.device)).getStatus()); + JSONObject desired = socket.take("DESIRED_STATE").getJSONObject("payload"); + assertTrue(desired.getBoolean("dispatchPaused")); + socket.apply(desired.getLong("revision"), 1); + cluster.serviceA.requireRealtimeReady(cluster.device); + + // A node that stopped without clearing its row is forgotten when the lease runs out; the socket's own node still answers for itself. + cluster.update("UPDATE printing_device SET realtime_connection_expires_at=? WHERE id=?", System.currentTimeMillis() - 1, cluster.device); + assertEquals(409, assertThrows(PrintingException.class, () -> cluster.serviceA.requireRealtimeReady(cluster.device)).getStatus()); + cluster.serviceB.requireRealtimeReady(cluster.device); + socket.send("PING", "renew-lease", new JSONObject()); + socket.take("PONG"); + await(() -> cluster.deviceRow().getLong("realtime_connection_expires_at") > System.currentTimeMillis()); + cluster.serviceA.requireRealtimeReady(cluster.device); + } + awaitReleased(cluster); + assertEquals(409, assertThrows(PrintingException.class, () -> cluster.serviceA.requireRealtimeReady(cluster.device)).getStatus()); + assertEquals(409, assertThrows(PrintingException.class, () -> cluster.serviceB.requireRealtimeReady(cluster.device)).getStatus()); + } + } + + + /** While the shared bus loses every message, the device's next heartbeat finds the work; an idle heartbeat stays silent. */ + @Test + void workCommittedWhileTheBusIsDownIsFoundOnTheNextHeartbeat() throws Exception + { + Nodes cluster = start("bus-down"); + try(HttpClient http = HttpClient.newHttpClient()) + { + String cookie = login(http, cluster.baseA); + try(PrintingSocketFixture socket = PrintingSocketFixture.connect(http, cluster.baseB, cookie, cluster.device, cluster.proof)) + { + socket.take("JOB_AVAILABLE"); + socket.send("PING", "idle", new JSONObject()); + socket.take("PONG"); + assertTrue(drain(socket).isEmpty(), "A heartbeat with nothing pending sends nothing but its PONG"); + + for(SwitchableBus.Mode mode : List.of(SwitchableBus.Mode.DROP, SwitchableBus.Mode.THROW)) + { + SwitchableBus.mode = mode; + long failures = cluster.serviceA.getRealtimeBus().getForwardFailures(); + String job = cluster.serviceA.submit(request(cluster.device, "bus-" + mode)).getString("jobId"); + assertTrue(drain(socket).isEmpty(), "The wakeup was lost with the bus (" + mode + ")"); + assertEquals("QUEUED", cluster.serviceB.getJob(job).getString("status")); + if(mode == SwitchableBus.Mode.THROW) + { + await(() -> cluster.serviceA.getRealtimeBus().getForwardFailures() > failures); + } + socket.send("PING", "heartbeat-" + mode, new JSONObject()); + socket.take("PONG"); + socket.take("JOB_AVAILABLE"); + assertEquals(job, json(send(http, cluster.baseA, "/device/jobs/claim", "POST", new JSONObject(), cookie, cluster), 200).getJSONObject("job").getString("jobId")); + json(send(http, cluster.baseA, "/device/jobs/" + job + "/events", "POST", new JSONObject().put("eventId", "received-" + mode).put("status", "RECEIVED"), cookie, cluster), 200); + socket.received.clear(); + socket.send("PING", "after-" + mode, new JSONObject()); + socket.take("PONG"); + assertTrue(drain(socket).isEmpty(), "A received job is not announced again"); + } + + // A settings change committed on node A while the bus is down also arrives with the heartbeat. + SwitchableBus.mode = SwitchableBus.Mode.DROP; + JSONObject policy = cluster.serviceA.getRealtimeSettings(); + policy.remove("revision"); + cluster.serviceA.saveRealtimeSettings(policy.put("maxCopies", 97)); + assertTrue(drain(socket).isEmpty()); + socket.send("PING", "settings-heartbeat", new JSONObject()); + socket.take("PONG"); + assertEquals(1, socket.take("SETTINGS").getJSONObject("payload").getLong("revision")); + socket.apply(0, 1); + socket.take("JOB_AVAILABLE"); + + // When the transport reports a gap after it recovers, every connection reconciles at once, without waiting for a heartbeat. + String late = cluster.serviceA.submit(request(cluster.device, "before-recovery")).getString("jobId"); + assertTrue(drain(socket).isEmpty()); + SwitchableBus.mode = SwitchableBus.Mode.UP; + SwitchableBus.announceGap(); + socket.take("JOB_AVAILABLE"); + assertEquals(late, cluster.serviceB.claim(cluster.device).getJSONObject("job").getString("jobId")); + } + } + } + + + /** A device has one connection in the whole application: connecting through another node closes the first, with or without the bus. */ + @Test + void aConnectionThroughAnotherNodeReplacesTheFirst() throws Exception + { + Nodes cluster = start("replace"); + try(HttpClient http = HttpClient.newHttpClient()) + { + String cookie = login(http, cluster.baseA); + try(PrintingSocketFixture onA = PrintingSocketFixture.connect(http, cluster.baseA, cookie, cluster.device, cluster.proof)) + { + onA.take("JOB_AVAILABLE"); + String first = cluster.deviceRow().getString("realtime_connection_id"); + try(PrintingSocketFixture onB = PrintingSocketFixture.connect(http, cluster.baseB, cookie, cluster.device, cluster.proof)) + { + assertEquals(1008, onA.closed.get(10, TimeUnit.SECONDS), "The bus told node A its connection was replaced"); + onB.take("JOB_AVAILABLE"); + String second = cluster.deviceRow().getString("realtime_connection_id"); + assertFalse(first.equals(second)); + cluster.serviceA.requireRealtimeReady(cluster.device); + await(() -> second.equals(cluster.deviceRow().optString("realtime_connection_id"))); + + // With the bus down node B is not told. Its connection learns on its next frame that the row names another connection. + SwitchableBus.mode = SwitchableBus.Mode.DROP; + try(PrintingSocketFixture backOnA = PrintingSocketFixture.connect(http, cluster.baseA, cookie, cluster.device, cluster.proof)) + { + backOnA.take("JOB_AVAILABLE"); + assertFalse(onB.closed.isDone(), "Node B has not been told"); + onB.send("PING", "stale", new JSONObject()); + assertEquals(1008, onB.closed.get(10, TimeUnit.SECONDS)); + JSONObject row = cluster.deviceRow(); + assertFalse(second.equals(row.getString("realtime_connection_id")), "The stale connection's close did not clear its replacement's row"); + assertEquals(1, row.getInt("realtime_connection_ready")); + backOnA.send("PING", "current", new JSONObject()); + backOnA.take("PONG"); + } + } + } + } + } + + + /** A database failure during the upgrade is retryable; only a refused session or a denial asks for a person. */ + @Test + void upgradeFailuresFromTheDatabaseAreRetryable() + { + assertEquals(503, PrintingRealtimeRoutes.upgradeStatus(new SQLException("connection refused"))); + assertEquals(503, PrintingRealtimeRoutes.upgradeStatus(new QException("lookup failed", new SQLException("pool exhausted")))); + assertEquals(503, PrintingRealtimeRoutes.upgradeStatus(new QAuthenticationException("session lookup failed", new SQLException("failover")))); + assertEquals(401, PrintingRealtimeRoutes.upgradeStatus(new QAuthenticationException("Session not found"))); + assertEquals(403, PrintingRealtimeRoutes.upgradeStatus(new QPermissionDeniedException("denied"))); + assertEquals(403, PrintingRealtimeRoutes.upgradeStatus(new PrintingException(404, "Device not found"))); + assertEquals(403, PrintingRealtimeRoutes.upgradeStatus(new IllegalArgumentException("bad origin"))); + } + + + /** Start node A (QQQ server) and node B (socket only) on one database and one application bus, with one enrolled device and an approved queue. */ + private Nodes start(String name) throws Exception + { + String jdbc = "jdbc:h2:file:" + directory.resolve(name).toString().replace('\\', '/'); + PrintingLocalHost host = new SharedBusHost(new PrintingLocalSettings(jdbc, "sa", "", "demo", 0, "synthetic-management-token-long-enough", "operator", "synthetic-realtime-password-1234")); + AtomicReference app = new AtomicReference<>(); + QApplicationLauncher launcher = QApplicationLauncher.run(host, new QApplicationLauncherConfig().withRegisterShutdownHook(false).withServerCustomizer(server -> + { + host.configure(server, false); + server.withJavalinConfigurationCustomizer(app::set); + })); + PrintingService serviceA = Objects.requireNonNull(launcher.getQInstance().getQBits().values().stream().map(value -> value.getConfig()).filter(PrintingQBitConfig.class::isInstance) + .map(PrintingQBitConfig.class::cast).findFirst().orElseThrow().getService(), "The printing service was not configured"); + DriverDataSource source = new DriverDataSource(jdbc, "sa", ""); + PrintingService serviceB = new PrintingService(source, "demo"); + PrintingRealtimeRoutes routesB = new PrintingRealtimeRoutes(serviceB); + routesB.setQInstance(launcher.getQInstance()); + Javalin other = Javalin.create(config -> + { + config.jetty.host = "127.0.0.1"; + routesB.acceptJavalinConfig(config); + }).start(0); + JSONObject enrolled = serviceA.enrollForUser(serviceA.createEnrollment("default", "Multi-node fixture").put("deviceName", "Multi-node fixture").put("agentVersion", "test"), "operator"); + String device = enrolled.getString("deviceId"); + serviceA.inventory(device, new JSONArray().put(new JSONObject().put("printerId", "queue").put("name", "Synthetic queue").put("status", "READY").put("formats", new JSONArray().put("ZPL")))); + serviceA.enablePrinter(device, "queue", true); + nodes = new Nodes(launcher, other, serviceA, serviceB, source, "http://127.0.0.1:" + Objects.requireNonNull(app.get(), "Test HTTP server was not started").port(), + "http://127.0.0.1:" + other.port(), device, enrolled.getString("deviceToken")); + return nodes; + } + + + /** The two nodes, their services and the one device they share. */ + private record Nodes(QApplicationLauncher launcher, Javalin other, PrintingService serviceA, PrintingService serviceB, DriverDataSource source, String baseA, String baseB, String device, String proof) + { + /** Read the device row both nodes consult. */ + JSONObject deviceRow() throws SQLException + { + try(Connection connection = source.getConnection()) + { + return PrintingService.query(connection, "SELECT realtime_connection_id, realtime_connection_node, realtime_connection_ready, realtime_connection_expires_at FROM printing_device WHERE id=?", device).getFirst(); + } + } + + + /** Count rows for an assertion. */ + long count(String sql, Object... values) throws SQLException + { + try(Connection connection = source.getConnection()) + { + return ((Number) PrintingService.query(connection, sql, values).getFirst().toMap().values().iterator().next()).longValue(); + } + } + + + /** Simulate what a stopped node leaves behind. */ + void update(String sql, Object... values) throws SQLException + { + try(Connection connection = source.getConnection()) + { + PrintingService.update(connection, sql, values); + } + } + + + /** Node B first, then the QQQ server. */ + void close() + { + other.stop(); + launcher.stop(); + } + } + + + /** The same local host, with the application's realtime bus replaced by one the test can switch off. */ + private static final class SharedBusHost extends PrintingLocalHost + { + /** Same settings as the ordinary local host. */ + SharedBusHost(PrintingLocalSettings settings) throws Exception + { + super(settings); + } + + + @Override + public QInstance defineQInstance() throws QException + { + QInstance instance = super.defineQInstance(); + instance.withSupplementalMetaData(RealtimeMetaData.of(instance).withBus(new QCodeReference(SwitchableBus.class))); + return instance; + } + } + + + /*************************************************************************** + ** Stands in for the shared transport (Redis in production): it delivers a + ** published message to every subscriber, the publisher included, and can + ** lose messages silently or fail, as an unreachable transport does. + ***************************************************************************/ + public static final class SwitchableBus implements RealtimeBusInterface + { + /** What the transport does with a published message. */ + enum Mode + { + UP, DROP, THROW + } + + static final AtomicInteger PUBLISHED = new AtomicInteger(); + private static final List> LISTENERS = new CopyOnWriteArrayList<>(); + static volatile Mode mode = Mode.UP; + + + /** Forget the previous test's subscribers and outage. */ + static void reset() + { + LISTENERS.clear(); + PUBLISHED.set(0); + mode = Mode.UP; + } + + + /** What a real transport emits locally after it reconnects or overflows. */ + static void announceGap() + { + LISTENERS.forEach(listener -> listener.accept(new RealtimeMessage().withChannel("instance").withEvent("resync"))); + } + + + @Override + public void publish(RealtimeMessage message) + { + if(mode == Mode.THROW) + { + throw new IllegalStateException("The shared bus is unreachable"); + } + if(mode == Mode.UP) + { + PUBLISHED.incrementAndGet(); + LISTENERS.forEach(listener -> listener.accept(message)); + } + } + + + @Override + public AutoCloseable subscribe(Consumer listener) + { + LISTENERS.add(listener); + return () -> LISTENERS.remove(listener); + } + } + + + /** Collect whatever the server pushes in a short bounded window; this is an absence check, not a polling loop. */ + private static List drain(PrintingSocketFixture socket) throws Exception + { + List frames = new ArrayList<>(); + JSONObject next = socket.received.poll(600, TimeUnit.MILLISECONDS); + while(next != null) + { + frames.add(next); + next = socket.received.poll(200, TimeUnit.MILLISECONDS); + } + return frames; + } + + + /** A condition another thread completes shortly. */ + private static void await(Condition condition) throws Exception + { + long end = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); + while(!condition.met()) + { + assertTrue(System.nanoTime() < end, "Condition was not met in time"); + Thread.sleep(20); + } + } + + + /** The server clears the device row when it sees the socket close. */ + private static void awaitReleased(Nodes cluster) throws Exception + { + await(() -> cluster.deviceRow().isNull("realtime_connection_id")); + assertNull(cluster.deviceRow().opt("realtime_connection_node")); + assertEquals(0, cluster.deviceRow().getInt("realtime_connection_ready")); + } + + + /** A check that may read the database. */ + @FunctionalInterface + private interface Condition + { + /** Whether the awaited state has been reached. */ + boolean met() throws Exception; + } + + + /** A real sign-in supplies only its own temporary test session, valid on both nodes because sessions are rows. */ + private static String login(HttpClient client, String base) throws Exception + { + HttpRequest request = HttpRequest.newBuilder(URI.create(base + "/qqq/v1/manageSession")).header("Origin", base) + .header("Authorization", "Basic " + Base64.getEncoder().encodeToString("operator:synthetic-realtime-password-1234".getBytes(StandardCharsets.UTF_8))) + .header("Content-Type", "application/json").POST(HttpRequest.BodyPublishers.ofString("{}")).build(); + HttpResponse response = client.send(request, HttpResponse.BodyHandlers.ofString()); + assertEquals(200, response.statusCode()); + return response.headers().allValues("set-cookie").stream().filter(value -> value.startsWith("sessionUUID=")).findFirst().orElseThrow().split(";", 2)[0]; + } + + + /** A native device request as the Windows client sends it: session cookie plus device proof. */ + private static HttpResponse send(HttpClient client, String base, String path, String method, @Nullable JSONObject body, String cookie, Nodes cluster) throws Exception + { + HttpRequest.Builder request = HttpRequest.newBuilder(URI.create(base + PREFIX + path)).header("Origin", base).header("X-QQQ-Client", "api/qqq-print-agent") + .header("Cookie", cookie).header(PrintingAgentAuthorization.DEVICE_HEADER, cluster.device).header(PrintingAgentAuthorization.TOKEN_HEADER, cluster.proof); + request.method(method, body == null ? HttpRequest.BodyPublishers.noBody() : HttpRequest.BodyPublishers.ofString(body.toString())); + if(body != null) + { + request.header("Content-Type", "application/json"); + } + return client.send(request.build(), HttpResponse.BodyHandlers.ofString()); + } + + + /** Require a status and parse the body. */ + private static JSONObject json(HttpResponse response, int status) + { + assertEquals(status, response.statusCode(), response.body()); + assertNotNull(response.body()); + return new JSONObject(response.body()); + } + + + /** Stable synthetic submission identity; no physical printer exists in this fixture. */ + private static JSONObject request(String device, String key) + { + return new JSONObject().put("deviceId", device).put("printerId", "queue").put("title", "Synthetic multi-node") + .put("format", "ZPL").put("contentBase64", Base64.getEncoder().encodeToString("^XA^XZ".getBytes(StandardCharsets.US_ASCII))).put("idempotencyKey", key); + } + + + /** Native actions retain their existing temporal replay boundary. */ + private static JSONObject action(String name) + { + return new JSONObject().put("action", name).put("requestId", java.util.UUID.randomUUID().toString()).put("issuedAt", java.time.Instant.now().toString()); + } +} diff --git a/src/test/java/com/kingsrook/qbits/printing/PrintingRealtimeBusTest.java b/src/test/java/com/kingsrook/qbits/printing/PrintingRealtimeBusTest.java new file mode 100644 index 0000000..cc95d79 --- /dev/null +++ b/src/test/java/com/kingsrook/qbits/printing/PrintingRealtimeBusTest.java @@ -0,0 +1,211 @@ +/* + * QQQ - Low-code Application Framework for Engineers. + * Copyright (C) 2021-2025. Kingsrook, LLC + * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States + * contact@kingsrook.com + * https://github.com/Kingsrook/ + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.kingsrook.qbits.printing; + + +import java.io.Serializable; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Consumer; +import com.kingsrook.qqq.backend.core.context.QContext; +import com.kingsrook.qqq.backend.core.model.metadata.QInstance; +import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference; +import com.kingsrook.qqq.backend.core.realtime.RealtimeBus; +import com.kingsrook.qqq.backend.core.realtime.RealtimeBusInterface; +import com.kingsrook.qqq.backend.core.realtime.RealtimeMessage; +import com.kingsrook.qqq.backend.core.realtime.RealtimeMetaData; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + + +/** The notification bridge joined to an application's QQQ realtime bus: delivery, echo suppression, foreign traffic and a slow transport. */ +class PrintingRealtimeBusTest +{ + /** The shared fixture leaves a native context on the thread. */ + @AfterEach + void clearContext() + { + QContext.clear(); + } + + + /** Two buses joined to one application each hear a wakeup once, whichever of them published it. */ + @Test + void wakeupsCrossJoinedBusesExactlyOnce() throws Exception + { + QInstance instance = PrintingProcessesTest.fixture().instance(); + PrintingRealtimeBus first = new PrintingRealtimeBus(); + PrintingRealtimeBus second = new PrintingRealtimeBus(); + List firstHeard = new CopyOnWriteArrayList<>(); + List secondHeard = new CopyOnWriteArrayList<>(); + AtomicInteger gaps = new AtomicInteger(); + first.subscribe((tenant, device) -> firstHeard.add(tenant + "/" + device)); + second.subscribe((tenant, device) -> secondHeard.add(tenant + "/" + device)); + second.subscribeGaps(gaps::incrementAndGet); + first.publish("tenant", "before-join"); + assertEquals(List.of("tenant/before-join"), firstHeard, "An unjoined bus still serves its own JVM"); + AutoCloseable joined = first.join(instance); + AutoCloseable joinedAgain = first.join(instance); + AutoCloseable other = second.join(instance); + try + { + first.publish("tenant", "device"); + first.publish("tenant", null); + await(() -> secondHeard.size() == 2); + assertEquals(List.of("tenant/device", "tenant/null"), secondHeard); + second.publish("tenant", "reply"); + await(() -> firstHeard.size() == 4); + // The transport echoes every message to its publisher; the publisher already delivered it locally and must not hear it twice. + Thread.sleep(200); + assertEquals(List.of("tenant/before-join", "tenant/device", "tenant/null", "tenant/reply"), firstHeard); + assertEquals(List.of("tenant/device", "tenant/null", "tenant/reply"), secondHeard); + + // Other traffic on the application's bus, and malformed printing messages, are not wakeups. + RealtimeBusInterface transport = RealtimeBus.get(instance); + transport.publish(new RealtimeMessage().withChannel("table:printing_job").withEvent("record.updated")); + transport.publish(new RealtimeMessage().withChannel(PrintingRealtimeBus.SHARED_CHANNEL).withEvent("other")); + transport.publish(message(Map.of("tenant", "tenant"))); + transport.publish(message(Map.of("origin", "elsewhere", "tenant", " "))); + transport.publish(message(Map.of("origin", "elsewhere"))); + assertEquals(4, firstHeard.size()); + assertEquals(3, secondHeard.size()); + transport.publish(message(Map.of("origin", "elsewhere", "tenant", "tenant", "device", "external"))); + assertEquals("tenant/external", firstHeard.getLast()); + assertEquals("tenant/external", secondHeard.getLast()); + + // The transport's own lost-message hint reaches gap listeners, not wakeup listeners. + transport.publish(new RealtimeMessage().withChannel("instance").withEvent("resync")); + assertEquals(1, gaps.get()); + assertEquals(5, firstHeard.size()); + + // Leaving is counted: the link stays while one user remains, and closing a handle twice counts once. + joinedAgain.close(); + joinedAgain.close(); + first.publish("tenant", "still-joined"); + await(() -> secondHeard.contains("tenant/still-joined")); + joined.close(); + first.publish("tenant", "after-leaving"); + transport.publish(message(Map.of("origin", "elsewhere", "tenant", "tenant", "device", "to-the-joined"))); + assertEquals("tenant/after-leaving", firstHeard.getLast(), "A bus that left hears only itself"); + assertEquals("tenant/to-the-joined", secondHeard.getLast()); + assertEquals(0, first.getForwardFailures()); + } + finally + { + other.close(); + RealtimeBus.close(instance); + } + } + + + /** The committing caller never waits for the transport, and wakeups the transport cannot take are counted, not thrown. */ + @Test + void aBlockedOrFailingTransportNeverDelaysThePublisher() throws Exception + { + QInstance instance = PrintingProcessesTest.fixture().instance(); + instance.withSupplementalMetaData(RealtimeMetaData.of(instance).withBus(new QCodeReference(BlockedBus.class))); + BlockedBus.release = new CountDownLatch(1); + PrintingRealtimeBus bus = new PrintingRealtimeBus(); + AtomicInteger heard = new AtomicInteger(); + bus.subscribe((tenant, device) -> heard.incrementAndGet()); + AutoCloseable joined = bus.join(instance); + try + { + long started = System.nanoTime(); + for(int index = 0; index < 1200; index++) + { + bus.publish("tenant", "device-" + index); + } + assertTrue(TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - started) < 5000, "Publishing waited for the transport"); + assertEquals(1200, heard.get(), "Every wakeup reached this JVM's listeners"); + assertTrue(bus.getForwardFailures() >= 1200 - 1024 - 1, "Wakeups beyond the bounded queue are counted as not forwarded"); + BlockedBus.release.countDown(); + long counted = bus.getForwardFailures(); + await(() -> BlockedBus.FAILED.get() > 0); + await(() -> bus.getForwardFailures() > counted); + } + finally + { + BlockedBus.release.countDown(); + joined.close(); + RealtimeBus.close(instance); + } + } + + + /** A transport that blocks until released and then fails every publish. */ + public static final class BlockedBus implements RealtimeBusInterface + { + static volatile CountDownLatch release = new CountDownLatch(0); + static final AtomicInteger FAILED = new AtomicInteger(); + + + @Override + public void publish(RealtimeMessage message) + { + try + { + release.await(); + } + catch(InterruptedException interrupted) + { + Thread.currentThread().interrupt(); + } + FAILED.incrementAndGet(); + throw new IllegalStateException("The shared bus is unreachable"); + } + + + @Override + public AutoCloseable subscribe(Consumer listener) + { + return () -> + { + }; + } + } + + + /** A printing wakeup as another node's bus would publish it. */ + private static RealtimeMessage message(Map values) + { + Map data = new java.util.LinkedHashMap<>(values); + return new RealtimeMessage().withChannel(PrintingRealtimeBus.SHARED_CHANNEL).withEvent(PrintingRealtimeBus.SHARED_EVENT).withData(data); + } + + + /** A condition the forwarding thread completes shortly. */ + private static void await(java.util.function.BooleanSupplier condition) throws InterruptedException + { + long end = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); + while(!condition.getAsBoolean()) + { + assertTrue(System.nanoTime() < end, "Condition was not met in time"); + Thread.sleep(10); + } + } +}