fix: bound print resource admission and inventory history
Co-authored-by: Codex (GPT-6) <noreply@openai.com> Signed-off-by: James Maes <james@kof22.com>
This commit is contained in:
12 files changed
+1184
-49
No files matched your search
+2
-1
@@ -8,7 +8,8 @@
|
||||
- Add admin retention settings with a 365-day default, reviewed cleanup and bounded scheduled maintenance; preserve job history and uncertain work.
|
||||
- Use named facility choices during enrollment and revalidate them under the current user's tenant/record permissions.
|
||||
- Require release security/CodeQL gates, verified artifact promotion and build provenance; report the real packaged QBit version.
|
||||
- Full local verification passed 53 tests with zero failures/errors and Checkstyle violations. Hosted checks remain pending; this entry does not claim release or live storage migration.
|
||||
- Enforce configurable atomic queue/storage/rate budgets and cumulative printer limits, with bounded management pagination and stable replay at capacity.
|
||||
- Full local verification passed 62 tests with zero failures/errors/skips and Checkstyle violations. Final hosted checks remain pending; this entry does not claim release or live storage migration.
|
||||
|
||||
## 0.1.3-rc.1 — 2026-10-30
|
||||
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
# Server printing audit
|
||||
|
||||
Target: 0.1.4-rc.1. Local implementation verification passed **53 tests, zero failures/errors and zero Checkstyle violations**. Hosted verification and the independent baseline security scan remain pending. No live database, object storage, enrollment or printer was changed.
|
||||
Target: 0.1.4-rc.1. Final local implementation verification passed **62 tests, zero failures/errors/skips and zero Checkstyle violations**. The independent baseline scan and source remediation review are complete. Earlier audit and CI guard commits passed hosted checks; final remediation commits require fresh verification. No live database, object storage, enrollment or printer was changed.
|
||||
|
||||
## Corrected behavior
|
||||
|
||||
@@ -12,6 +12,8 @@ Target: 0.1.4-rc.1. Local implementation verification passed **53 tests, zero fa
|
||||
- Durable upload intents and exact orphan cleanup account for failed uploads/queue commits. Cleanup commits each confirmed document removal independently. A renewable lease prevents normal overlapping maintenance; all work reuses a single connection and passes real one-/two-connection pool concurrency regressions.
|
||||
- Native administration exposes the default 365-day policy, read-only storage scope, backlog and last cleanup result. A host scheduler runs bounded batches every five minutes. Age starts at server-observed successful completion; active, failed, uncertain and unknown-completion jobs remain protected.
|
||||
- Release publication requires build/security/CodeQL checks, verified artifact promotion and build provenance. Packaged QBit metadata uses the JAR's real Implementation-Version.
|
||||
- Shared submission admission enforces configurable atomic device/tenant limits for outstanding work, retained bytes and hourly allocations. Durable uploads remain charged until confirmed cleanup; matching receipts replay even at capacity. Legacy unknown uploads conservatively reserve 20 MiB.
|
||||
- Cumulative printer limits prevent rotating identities from growing history indefinitely. Management pages remain bounded, with optional opaque cursors; approval, metadata and job references survive rejected snapshots. A narrow index avoids PostgreSQL limits on wide Unicode names.
|
||||
|
||||
## Evidence and migration
|
||||
|
||||
@@ -19,8 +21,10 @@ The gate exercises real H2 persistence and bounded pools, QQQ StorageAction with
|
||||
|
||||
Schema initialization adds consumed-enrollment receipts, storage references/content availability, completion timestamps, retention settings/lease and upload intents. Existing inline bytes are retained; completion dates are backfilled only from durable successful events. No automatic inline-to-S3 copy or live purge occurs. Schema changes should be qualified against a backed-up deployment database before rollout.
|
||||
|
||||
Admission accounting also backfills retained bytes and recent allocations. Stop older writers before this migration; mixed-version writers bypass accounting and are unsupported. Regression cases cover concurrent tenant admissions, retained terminal payloads, replay at capacity, failed cleanup, unknown legacy uploads, rotating inventory and long-name pagination. See [operations](OPERATIONS.md) for configurable limits and recovery.
|
||||
|
||||
## Remaining boundaries
|
||||
|
||||
Live S3 endpoints, IAM/bucket policies, external bucket users, PostgreSQL, multi-node lease-expiry behavior, production scheduler execution, remote TLS deployment and sustained 10,000-print/day throughput are not qualified by these tests. SQL and object deletion are not one transaction; crashes between them require idempotent recovery. S3 versioning and unfinished multipart parts have separate lifecycle semantics. See [storage and retention](STORAGE-AND-RETENTION.md).
|
||||
|
||||
The independent deep scan covers the frozen pre-audit baseline, not the subsequently added storage feature. Two implementation agents independently reviewed the new storage boundaries and added regressions, but that does not substitute for a completed security scan or prove absence of defects. No numerical quality score is assigned.
|
||||
The completed independent deep scan covers the frozen pre-audit baseline, not the subsequently added storage feature. Its two server findings were remediated and received a fresh independent bypass/regression review; that review's wide-index compatibility finding was corrected before the final gate. Separate review and regressions cover the new storage boundaries. These results do not prove absence of defects or qualify the live deployment boundaries above. No numerical quality score is assigned.
|
||||
@@ -26,6 +26,26 @@ Use a stable business idempotency key for retries of the same request. A changed
|
||||
|
||||
The pending 0.1.4 implementation stores new documents through the host's configured dedicated QQQ S3 storage table; existing inline database content remains readable. Administration defaults to 365 days after server-observed successful completion, with bounded scheduled cleanup when a host scheduler is configured. Active, failed, uncertain and unknown-completion legacy jobs remain protected. See [storage and retention](STORAGE-AND-RETENTION.md) for bucket separation, recovery and versioning details. No live storage migration or cleanup is implied by a source upgrade.
|
||||
|
||||
## Admission capacity and inventory history
|
||||
|
||||
The host configures resource ceilings through `PrintingQBitConfig.withAdmissionLimits(PrintingAdmissionLimits)`. The producer passes these limits to every submission path, including native processes, direct Java submissions, connector test prints and explicit retries. Retention remains a separate administrator setting, defaulting to 365 days after successful completion.
|
||||
|
||||
| Limit | Per device | Per tenant |
|
||||
| --- | ---: | ---: |
|
||||
| Outstanding jobs and upload reservations | 10,000 | 100,000 |
|
||||
| Outstanding decoded document bytes | 10 GiB | 100 GiB |
|
||||
| Retained decoded document bytes, including terminal jobs and upload reservations | 2 TiB | 20 TiB |
|
||||
| New admissions in a rolling hour | 2,000 | 20,000 |
|
||||
| Cumulative printer identities | 1,000 | 20,000 |
|
||||
|
||||
`PrintingAdmissionLimits.defaults()` returns these defaults. To override them, construct the record with device/tenant pairs in this order: outstanding jobs, outstanding bytes, retained bytes, hourly admissions, cumulative printers. Values must be positive. Use the same configuration on every QQQ node serving a tenant. Size retained capacity for actual document sizes and the chosen retention period; 10,000 jobs/day is a sizing target, not a measured hardware throughput result. These are logical application byte limits, not an S3 billing limit: object versions, multipart fragments and provider overhead require their own deployment monitoring and lifecycle policy.
|
||||
|
||||
Admission is serialized per tenant in SQL before storage writes. Pending and failed upload reservations stay charged until exact-object cleanup succeeds; failed admissions that reached a committed reservation also consume the hourly allowance. Legacy reservations with unknown length conservatively reserve the maximum 20 MiB each, against the tenant when their device is unknown. Existing payloads are backfilled without removal. Apply this additive accounting migration with older application writers stopped; mixed-version writers or direct SQL changes bypass accounting and are unsupported.
|
||||
|
||||
HTTP 429 means capacity, the rolling-hour allowance, cumulative inventory history, or an existing upload reservation prevents a new allocation. The response explains the category. Reuse the same idempotency/request identity when retrying: a completed matching replay returns its existing job even at capacity or after the original retry source expires. Wait for outstanding work or the hourly window, investigate failed cleanup/storage access, or have the host administrator adjust a justified capacity limit. Retention only removes eligible successful documents; it does not erase failed, uncertain or active work to make space.
|
||||
|
||||
Inventory snapshots remain limited to 200 entries. Rotating IDs cannot bypass cumulative history limits, and rejected snapshots leave approval, availability and job references intact. Existing IDs remain updateable when a legacy database already exceeds a limit. Management `GET /printing/v1/admin/devices/{deviceId}/printers` accepts `limit` (1–200, default 200) and opaque `before` from the preceding response's `nextCursor`. The response is `{ "printers": [...], "nextCursor": null | "..." }`, ordered by stored name and row identity. The existing Java `printers()` method returns the first bounded array; use `printersPage()` for additional history. Native QQQ tables retain their own backend paging.
|
||||
|
||||
## Common failures
|
||||
|
||||
| Symptom | Check |
|
||||
|
||||
@@ -1,12 +1,14 @@
|
||||
# Session state — current checkpoint 2026-10-30
|
||||
# Session state — 2026-10-30
|
||||
|
||||
## Deep audit in progress — pending 0.1.4-rc.1
|
||||
## Audit candidate 0.1.4-rc.1
|
||||
|
||||
Final local server verification passed **53 tests, zero failures/errors and zero Checkstyle violations**, including named facility selectors and current-user tenant authorization. The JAR manifest reports 0.1.4-rc.1. See [audit evidence](AUDIT-2026-10-30.md). Root is preparing the draft PR; no merge, release or live migration. All three printing repositories now enforce required checks even for administrators, explicitly approved by the user. Master coordination is in `R:\agent-comms\qqq-cloud-printing.md`; Record Files owns originals, printing owns delivery copies. The shared immutable-version reader/async source-authorization contract remains pending. Legacy durable outbox byte payloads are outside printing's S3 retention scope.
|
||||
Final isolated verification passed **62 tests, zero failures/errors/skips and zero Checkstyle violations**. The independent baseline scan's two server findings were remediated: configurable atomic tenant/device admission covers outstanding work, retained bytes and hourly allocations; cumulative printer limits and cursor pagination bound inventory. A fresh reviewer found a PostgreSQL wide-index regression, corrected before the final gate. Durable upload accounting, legacy unknown reservations, stable replay and single-connection cleanup have regression coverage.
|
||||
|
||||
Source fixes address native process replay, strict print options/inventory validation, duplicate-risk preservation and event bounds. The user requires admin-configurable document retention, default **365 days**, scheduled cleanup and proper S3 bucket separation. Implementation uses existing QQQ StorageAction/S3 configuration with a dedicated printing bucket, tenant namespaces and immutable per-job storage bindings; exact cleanup scope and cross-tenant isolation require regression evidence before release. Active/uncertain jobs and history/receipts remain protected. Existing inline documents remain readable. No live storage configuration, purge or database migration has run.
|
||||
Earlier fixes add replay-safe native processes, validated facility selectors, strict wire fields, duplicate-risk preservation, dedicated QQQ S3 storage and native admin retention with a **365-day default**. Active, failed and uncertain work remains protected. Stop older writers before admission-accounting migration; existing content is preserved. See [audit evidence](AUDIT-2026-10-30.md), [operations](OPERATIONS.md) and [storage](STORAGE-AND-RETENTION.md).
|
||||
|
||||
Root owns publication and build/release gates; server/native-metadata agents own source and tests. Source, hosted provenance and independent security results remain pending. The installed Windows baseline is still 0.1.7; unsigned prereleases are explicitly accepted by the user. Do not infer deployed-cloud or new hardware acceptance from these source changes.
|
||||
Draft PR #10 contains earlier audit/CI guard commits that passed hosted checks; final remediation commits need fresh checks. No live database migration, S3 provisioning, purge, merge, release or host deployment has occurred. Real PostgreSQL/S3, remote host authentication, multi-node operation and sustained physical throughput remain unqualified.
|
||||
|
||||
Master coordination at `R:\agent-comms\qqq-cloud-printing.md` is acknowledged. Record Files owns indefinitely retained originals; printing owns its dedicated delivery copies. Their compiled source-reader/actor authorization contract remains pending. Standalone byte submission continues. Installed Windows baseline remains 0.1.7; unsigned prereleases are user-approved.
|
||||
|
||||
## Branded diagnostic documents — v0.1.3-rc.1 verified and published
|
||||
|
||||
|
||||
@@ -4,6 +4,7 @@
|
||||
- [x] Implement admin-configurable document retention (one-year default), bounded scheduled cleanup and existing QQQ object-storage integration, retaining history/receipts and protecting active or uncertain output.
|
||||
- [ ] Integrate Record Files' authorized immutable-version reader after its contract is implemented; define async source revocation/correlation and eliminate host-outbox byte copies for that path.
|
||||
- [ ] Verify audit release gates and provenance; complete independent security scan and deployed-host qualification.
|
||||
- [x] Audit PR #10 at 1655d4e passes all hosted checks; download and verify JAR/POM artifacts. Tagged-release provenance, deep scan and deployment remain pending.
|
||||
|
||||
- [x] Versioned shared contract and HTTP happy-path test.
|
||||
- [x] Durable enrollment, device-scoped authentication, inventory and printer approval.
|
||||
@@ -50,3 +51,5 @@ The Windows gate, MSI rebuild, installed UI acceptance, release evidence and fre
|
||||
- [x] Correct the Zebra machine driver defaults from 4.09 × 2-inch Thermal Transfer to 4 × 6-inch Thermal Direct, restore/resume the printer, and confirm one fresh built-in PDF printed correctly. No application code fix was needed.
|
||||
- [x] Verify published Windows 0.1.7 assets; all three built-in physical paths are confirmed: laser PDF, Zebra ZPL and Zebra PDF.
|
||||
- [ ] Obtain explicit barcode scan and measured alignment confirmation; successful spooling is not physical acceptance.
|
||||
|
||||
- [ ] Complete independent review and deployment qualification after the five baseline security findings are remediated; current details are in SESSION-STATE.md.
|
||||
@@ -0,0 +1,289 @@
|
||||
/*
|
||||
* 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.sql.Connection;
|
||||
import java.sql.SQLException;
|
||||
import java.time.Instant;
|
||||
import java.util.Set;
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Tenant-serialized, payload-free resource accounting. Callers use one JDBC
|
||||
** connection and acquire this lock before device, document or inventory locks.
|
||||
*******************************************************************************/
|
||||
final class PrintingAdmission
|
||||
{
|
||||
static final long UNKNOWN_UPLOAD_BYTES = 20L * 1024 * 1024;
|
||||
private static final String ACTIVE = "('QUEUED','CLAIMED','RECEIVED','SUBMITTING')";
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Static SQL helpers only.
|
||||
***************************************************************************/
|
||||
private PrintingAdmission()
|
||||
{
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Add accounting without purging old payloads or assuming unknown bytes zero.
|
||||
***************************************************************************/
|
||||
static void initialize(Connection connection, String tenant) throws SQLException
|
||||
{
|
||||
for(String sql : new String[] {
|
||||
"ALTER TABLE printing_document_orphan ADD COLUMN IF NOT EXISTS device_id VARCHAR(36)",
|
||||
"ALTER TABLE printing_document_orphan ADD COLUMN IF NOT EXISTS content_length BIGINT",
|
||||
"ALTER TABLE printing_document_orphan ADD COLUMN IF NOT EXISTS idempotency_key VARCHAR(300)",
|
||||
"CREATE TABLE IF NOT EXISTS printing_resource_usage (tenant_id VARCHAR(200) NOT NULL, device_id VARCHAR(36) NOT NULL, retained_bytes BIGINT NOT NULL, PRIMARY KEY(tenant_id,device_id))",
|
||||
"CREATE TABLE IF NOT EXISTS printing_admission_event (id VARCHAR(36) PRIMARY KEY, tenant_id VARCHAR(200) NOT NULL, device_id VARCHAR(36), created_at VARCHAR(40) NOT NULL)",
|
||||
"CREATE INDEX IF NOT EXISTS printing_admission_tenant_time_idx ON printing_admission_event(tenant_id,created_at)",
|
||||
"CREATE INDEX IF NOT EXISTS printing_admission_device_time_idx ON printing_admission_event(tenant_id,device_id,created_at)",
|
||||
"CREATE INDEX IF NOT EXISTS printing_job_outstanding_tenant_idx ON printing_job(tenant_id,status,content_length)",
|
||||
"CREATE INDEX IF NOT EXISTS printing_job_outstanding_device_idx ON printing_job(tenant_id,device_id,status,content_length)",
|
||||
"CREATE INDEX IF NOT EXISTS printing_orphan_device_idx ON printing_document_orphan(tenant_id,device_id)",
|
||||
"CREATE INDEX IF NOT EXISTS printing_orphan_key_idx ON printing_document_orphan(tenant_id,idempotency_key)",
|
||||
"CREATE INDEX IF NOT EXISTS printing_printer_inventory_idx ON printing_printer(device_id,available)",
|
||||
"CREATE INDEX IF NOT EXISTS printing_printer_owner_idx ON printing_printer(tenant_id,device_id,id)"
|
||||
})
|
||||
{
|
||||
execute(connection, sql);
|
||||
}
|
||||
initializeTenant(connection, tenant);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Existing schema may serve newly configured tenants without another DDL run.
|
||||
***************************************************************************/
|
||||
static void initializeTenant(Connection connection, String tenant) throws SQLException
|
||||
{
|
||||
connection.setAutoCommit(false);
|
||||
if(number(connection, "SELECT COUNT(*) FROM printing_resource_usage WHERE tenant_id=? AND device_id=''", tenant) == 0)
|
||||
{
|
||||
try
|
||||
{
|
||||
execute(connection, "INSERT INTO printing_resource_usage VALUES (?, '', 0)", tenant);
|
||||
}
|
||||
catch(SQLException exception)
|
||||
{
|
||||
if(!"23505".equals(exception.getSQLState()))
|
||||
{
|
||||
throw exception;
|
||||
}
|
||||
connection.rollback();
|
||||
connection.setAutoCommit(true);
|
||||
return;
|
||||
}
|
||||
try
|
||||
{
|
||||
lock(connection, tenant);
|
||||
long bytes = number(connection, "SELECT COALESCE(SUM(content_length),0) FROM printing_job WHERE tenant_id=? AND content_available=1", tenant)
|
||||
+ orphanBytes(connection, tenant, null);
|
||||
execute(connection, "UPDATE printing_resource_usage SET retained_bytes=? WHERE tenant_id=? AND device_id=''", bytes, tenant);
|
||||
execute(connection, "INSERT INTO printing_resource_usage (tenant_id,device_id,retained_bytes) SELECT d.tenant_id,d.id,COALESCE((SELECT SUM(j.content_length) FROM printing_job j WHERE j.tenant_id=d.tenant_id AND j.device_id=d.id AND j.content_available=1),0)+COALESCE((SELECT SUM(COALESCE(o.content_length,20971520)) FROM printing_document_orphan o WHERE o.tenant_id=d.tenant_id AND o.device_id=d.id AND NOT EXISTS(SELECT 1 FROM printing_job j WHERE j.id=o.id)),0) FROM printing_device d WHERE d.tenant_id=?", tenant);
|
||||
String cutoff = Instant.now().minusSeconds(3600).toString();
|
||||
execute(connection, "INSERT INTO printing_admission_event SELECT id,tenant_id,device_id,created_at FROM printing_job WHERE tenant_id=? AND created_at>=?", tenant, cutoff);
|
||||
execute(connection, "INSERT INTO printing_admission_event SELECT o.id,o.tenant_id,o.device_id,o.created_at FROM printing_document_orphan o WHERE o.tenant_id=? AND o.created_at>=? AND NOT EXISTS(SELECT 1 FROM printing_admission_event a WHERE a.id=o.id)", tenant, cutoff);
|
||||
connection.commit();
|
||||
}
|
||||
catch(Exception exception)
|
||||
{
|
||||
connection.rollback();
|
||||
throw exception;
|
||||
}
|
||||
finally
|
||||
{
|
||||
connection.setAutoCommit(true);
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
connection.commit();
|
||||
connection.setAutoCommit(true);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Shared lock serializes capacity decisions across devices and processes.
|
||||
***************************************************************************/
|
||||
static void lock(Connection connection, String tenant) throws SQLException
|
||||
{
|
||||
try(var statement = connection.prepareStatement("SELECT retained_bytes FROM printing_resource_usage WHERE tenant_id=? AND device_id='' FOR UPDATE"))
|
||||
{
|
||||
statement.setString(1, tenant);
|
||||
try(var rows = statement.executeQuery())
|
||||
{
|
||||
if(!rows.next())
|
||||
{
|
||||
throw new SQLException("Printing admission accounting has not been initialized");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Admission precedes every upload/inline insert; rollback restores all usage.
|
||||
***************************************************************************/
|
||||
static void reserve(Connection connection, String tenant, String device, String id, int bytes, PrintingAdmissionLimits limits) throws SQLException
|
||||
{
|
||||
if(number(connection, "SELECT COUNT(*) FROM printing_resource_usage WHERE tenant_id=? AND device_id=?", tenant, device) == 0)
|
||||
{
|
||||
execute(connection, "INSERT INTO printing_resource_usage VALUES (?, ?, 0)", tenant, device);
|
||||
}
|
||||
String cutoff = Instant.now().minusSeconds(3600).toString();
|
||||
execute(connection, "DELETE FROM printing_admission_event WHERE tenant_id=? AND created_at<?", tenant, cutoff);
|
||||
checkScope(connection, tenant, null, bytes, limits.tenantOutstandingJobs(), limits.tenantOutstandingBytes(), limits.tenantRetainedBytes(), limits.tenantJobsPerHour(), cutoff);
|
||||
checkScope(connection, tenant, device, bytes, limits.deviceOutstandingJobs(), limits.deviceOutstandingBytes(), limits.deviceRetainedBytes(), limits.deviceJobsPerHour(), cutoff);
|
||||
execute(connection, "UPDATE printing_resource_usage SET retained_bytes=retained_bytes+? WHERE tenant_id=? AND device_id IN ('',?)", bytes, tenant, device);
|
||||
execute(connection, "INSERT INTO printing_admission_event VALUES (?,?,?,?)", id, tenant, device, Instant.now().toString());
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** SQL aggregates stay bounded by active capacity and the rolling hour index.
|
||||
***************************************************************************/
|
||||
private static void checkScope(Connection connection, String tenant, String device, int bytes, long jobLimit, long activeByteLimit, long retainedLimit, long hourlyLimit, String cutoff) throws SQLException
|
||||
{
|
||||
String predicate = "tenant_id=?" + (device == null ? "" : " AND device_id=?");
|
||||
String outstanding = "SELECT content_length FROM printing_job WHERE " + predicate + " AND status IN " + ACTIVE
|
||||
+ " UNION ALL SELECT COALESCE(content_length,20971520) FROM printing_document_orphan o WHERE " + predicate + " AND NOT EXISTS(SELECT 1 FROM printing_job j WHERE j.id=o.id)";
|
||||
Object[] bothScopes = device == null ? new Object[] {tenant, tenant} : new Object[] {tenant, device, tenant, device};
|
||||
long jobs = number(connection, "SELECT COUNT(*) FROM (" + outstanding + ") outstanding", bothScopes);
|
||||
long activeBytes = number(connection, "SELECT COALESCE(SUM(content_length),0) FROM (" + outstanding + ") outstanding", bothScopes);
|
||||
long retained = number(connection, "SELECT retained_bytes FROM printing_resource_usage WHERE tenant_id=? AND device_id=?", tenant, device == null ? "" : device);
|
||||
Object[] rateScope = device == null ? new Object[] {tenant, cutoff} : new Object[] {tenant, device, cutoff};
|
||||
long rate = number(connection, "SELECT COUNT(*) FROM printing_admission_event WHERE " + predicate + " AND created_at>=?", rateScope);
|
||||
String scopeName = device == null ? "tenant" : "device";
|
||||
if(jobs >= jobLimit || bytes > activeByteLimit || activeBytes > activeByteLimit - bytes)
|
||||
{
|
||||
throw new PrintingException(429, "The " + scopeName + " outstanding print-job capacity is full; wait for existing work or ask an administrator to adjust admission limits");
|
||||
}
|
||||
if(bytes > retainedLimit || retained > retainedLimit - bytes)
|
||||
{
|
||||
throw new PrintingException(429, "The " + scopeName + " retained print-document capacity is full; review retention and storage capacity before retrying");
|
||||
}
|
||||
if(rate >= hourlyLimit)
|
||||
{
|
||||
throw new PrintingException(429, "The " + scopeName + " hourly print-job admission limit is reached; retry later using the same request identity");
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Failed upload reservations stay charged until exact confirmed removal.
|
||||
***************************************************************************/
|
||||
private static long orphanBytes(Connection connection, String tenant, String device) throws SQLException
|
||||
{
|
||||
String sql = "SELECT COALESCE(SUM(COALESCE(content_length,20971520)),0) FROM printing_document_orphan o WHERE tenant_id=?"
|
||||
+ (device == null ? "" : " AND device_id=?") + " AND NOT EXISTS(SELECT 1 FROM printing_job j WHERE j.id=o.id)";
|
||||
return (device == null ? number(connection, sql, tenant) : number(connection, sql, tenant, device));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Release only after deletion; an inconsistent counter fails closed.
|
||||
***************************************************************************/
|
||||
static void release(Connection connection, String tenant, String device, long bytes) throws SQLException
|
||||
{
|
||||
int expected = device == null ? 1 : 2;
|
||||
String predicate = device == null ? "device_id=''" : "device_id IN ('',?)";
|
||||
Object[] values = device == null ? new Object[] {bytes, tenant, bytes} : new Object[] {bytes, tenant, device, bytes};
|
||||
if(execute(connection, "UPDATE printing_resource_usage SET retained_bytes=retained_bytes-? WHERE tenant_id=? AND " + predicate + " AND retained_bytes>=?", values) != expected)
|
||||
{
|
||||
throw new SQLException("Printing retained-byte accounting requires administrator reconciliation");
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Cumulative IDs cannot bypass limits by rotating successive snapshots.
|
||||
***************************************************************************/
|
||||
static void inventory(Connection connection, String tenant, String device, Set<String> incoming, PrintingAdmissionLimits limits) throws SQLException
|
||||
{
|
||||
int additions = 0;
|
||||
for(String id : incoming)
|
||||
{
|
||||
if(number(connection, "SELECT COUNT(*) FROM printing_printer WHERE tenant_id=? AND device_id=? AND printer_id=?", tenant, device, id) == 0)
|
||||
{
|
||||
additions++;
|
||||
}
|
||||
}
|
||||
if(additions > 0 && (number(connection, "SELECT COUNT(*) FROM printing_printer WHERE tenant_id=? AND device_id=?", tenant, device) > limits.devicePrinters() - additions
|
||||
|| number(connection, "SELECT COUNT(*) FROM printing_printer WHERE tenant_id=?", tenant) > limits.tenantPrinters() - additions))
|
||||
{
|
||||
throw new PrintingException(429, "Printer history capacity is full; existing printers remain available to update. Ask an administrator to review the cumulative inventory limit");
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Read one scalar; never materialize payloads or an unbounded row list.
|
||||
***************************************************************************/
|
||||
private static long number(Connection connection, String sql, Object... values) throws SQLException
|
||||
{
|
||||
try(var statement = connection.prepareStatement(sql))
|
||||
{
|
||||
for(int index = 0; index < values.length; index++)
|
||||
{
|
||||
statement.setObject(index + 1, values[index]);
|
||||
}
|
||||
try(var rows = statement.executeQuery())
|
||||
{
|
||||
if(!rows.next())
|
||||
{
|
||||
throw new SQLException("Missing printing resource accounting row");
|
||||
}
|
||||
return (rows.getLong(1));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** All identifiers are fixed SQL; only values come from callers.
|
||||
***************************************************************************/
|
||||
private static int execute(Connection connection, String sql, Object... values) throws SQLException
|
||||
{
|
||||
try(var statement = connection.prepareStatement(sql))
|
||||
{
|
||||
for(int index = 0; index < values.length; index++)
|
||||
{
|
||||
statement.setObject(index + 1, values[index]);
|
||||
}
|
||||
return (statement.executeUpdate());
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,54 @@
|
||||
/*
|
||||
* 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;
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Host-configurable logical resource ceilings, independent of retention age.
|
||||
*******************************************************************************/
|
||||
public record PrintingAdmissionLimits(int deviceOutstandingJobs, int tenantOutstandingJobs,
|
||||
long deviceOutstandingBytes, long tenantOutstandingBytes, long deviceRetainedBytes, long tenantRetainedBytes,
|
||||
int deviceJobsPerHour, int tenantJobsPerHour, int devicePrinters, int tenantPrinters)
|
||||
{
|
||||
/***************************************************************************
|
||||
** Never silently disable a security boundary through zero/negative limits.
|
||||
***************************************************************************/
|
||||
public PrintingAdmissionLimits
|
||||
{
|
||||
if(deviceOutstandingJobs < 1 || tenantOutstandingJobs < 1 || deviceOutstandingBytes < 1 || tenantOutstandingBytes < 1
|
||||
|| deviceRetainedBytes < 1 || tenantRetainedBytes < 1 || deviceJobsPerHour < 1 || tenantJobsPerHour < 1 || devicePrinters < 1 || tenantPrinters < 1)
|
||||
{
|
||||
throw new IllegalArgumentException("Printing admission limits must all be positive");
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Capacity must be sized for document sizes and the deployment's retention.
|
||||
***************************************************************************/
|
||||
public static PrintingAdmissionLimits defaults()
|
||||
{
|
||||
long gibibyte = 1024L * 1024 * 1024;
|
||||
return (new PrintingAdmissionLimits(10_000, 100_000, 10 * gibibyte, 100 * gibibyte,
|
||||
2048 * gibibyte, 20_480 * gibibyte, 2_000, 20_000, 1_000, 20_000));
|
||||
}
|
||||
}
|
||||
@@ -31,6 +31,7 @@ public class PrintingQBitConfig implements QBitConfig
|
||||
{
|
||||
private String backendName;
|
||||
private transient PrintingService service;
|
||||
private PrintingAdmissionLimits admissionLimits = PrintingAdmissionLimits.defaults();
|
||||
private PrintingRetentionPolicy retentionPolicy = new PrintingRetentionPolicy(true, 365, 200);
|
||||
private String storageTableName;
|
||||
private String schedulerName;
|
||||
@@ -38,6 +39,27 @@ public class PrintingQBitConfig implements QBitConfig
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Resource capacity is deployment-owned and independent of retention age.
|
||||
***************************************************************************/
|
||||
public PrintingAdmissionLimits getAdmissionLimits()
|
||||
{
|
||||
return (admissionLimits);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Positive tenant/device limits apply at the shared admission boundary.
|
||||
***************************************************************************/
|
||||
public PrintingQBitConfig withAdmissionLimits(PrintingAdmissionLimits value)
|
||||
{
|
||||
admissionLimits = java.util.Objects.requireNonNull(value);
|
||||
return (this);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Reuse the host's facility selector and its existing record security.
|
||||
***************************************************************************/
|
||||
|
||||
@@ -79,7 +79,7 @@ public class PrintingQBitProducer implements QBitProducer
|
||||
{
|
||||
throw new QException("The initial printing QBit supports the printing namespace only");
|
||||
}
|
||||
config.getService().withRetentionPolicy(config.getRetentionPolicy());
|
||||
config.getService().withRetentionPolicy(config.getRetentionPolicy()).withAdmissionLimits(config.getAdmissionLimits());
|
||||
config.getService().configureStorage(instance, config.getStorageTableName());
|
||||
instance.addQBit(new QBitMetaData().withGroupId("com.kingsrook.qbits").withArtifactId("qbit-printing").withVersion(version()).withNamespace("printing").withConfig(config));
|
||||
instance.addSecurityKeyType(new QSecurityKeyType().withName("printing_tenant"));
|
||||
|
||||
@@ -107,7 +107,8 @@ public class PrintingRoutes implements QJavalinRouteProviderInterface
|
||||
config.routes.get(BASE + "/admin/devices/{deviceId}/printers", context -> handle(context, () ->
|
||||
{
|
||||
admin(context);
|
||||
return new JSONObject().put("printers", service.printers(context.pathParam("deviceId")));
|
||||
String limit = context.queryParam("limit");
|
||||
return service.printersPage(context.pathParam("deviceId"), limit == null ? 200 : Integer.parseInt(limit), context.queryParam("before"));
|
||||
}));
|
||||
config.routes.put(BASE + "/admin/devices/{deviceId}/printers/{printerId}", context -> handle(context, () ->
|
||||
{
|
||||
|
||||
@@ -52,11 +52,23 @@ public class PrintingService
|
||||
private final DataSource dataSource;
|
||||
private final String tenantId;
|
||||
private PrintingDocumentStorage documentStorage;
|
||||
private PrintingAdmissionLimits admissionLimits = PrintingAdmissionLimits.defaults();
|
||||
private String storageTableName;
|
||||
private PrintingRetentionPolicy defaultRetentionPolicy = new PrintingRetentionPolicy(true, 365, 200);
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Deployment-owned limits apply equally to native, HTTP and outbox callers.
|
||||
***************************************************************************/
|
||||
public PrintingService withAdmissionLimits(PrintingAdmissionLimits limits)
|
||||
{
|
||||
admissionLimits = java.util.Objects.requireNonNull(limits);
|
||||
return (this);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Bind a host-owned storage table before exposing this service to requests.
|
||||
***************************************************************************/
|
||||
@@ -165,6 +177,7 @@ public class PrintingService
|
||||
statement.execute();
|
||||
}
|
||||
}
|
||||
PrintingAdmission.initialize(connection, tenantId);
|
||||
widenColumn(connection, "printing_device", "name", 1024);
|
||||
widenColumn(connection, "printing_enrollment", "device_name", 1024);
|
||||
widenColumn(connection, "printing_printer", "name", 1024);
|
||||
@@ -418,11 +431,21 @@ public class PrintingService
|
||||
}
|
||||
try(Connection connection = dataSource.getConnection())
|
||||
{
|
||||
requireDevice(connection, deviceId);
|
||||
PrintingAdmission.initializeTenant(connection, tenantId);
|
||||
connection.setAutoCommit(false);
|
||||
try
|
||||
{
|
||||
PrintingAdmission.lock(connection, tenantId);
|
||||
requireDevice(connection, deviceId);
|
||||
update(connection, "UPDATE printing_printer SET status='UNKNOWN', available=0 WHERE device_id=?", deviceId);
|
||||
query(connection, "SELECT id FROM printing_device WHERE id=? FOR UPDATE", deviceId);
|
||||
Set<String> incoming = new java.util.HashSet<>();
|
||||
for(int index = 0; index < printers.length(); index++)
|
||||
{
|
||||
incoming.add(required(printers.getJSONObject(index).optString("printerId"), "printerId"));
|
||||
}
|
||||
PrintingAdmission.inventory(connection, tenantId, deviceId, incoming, admissionLimits);
|
||||
update(connection, "UPDATE printing_printer SET status='UNKNOWN', available=0 WHERE device_id=? AND available=1", deviceId);
|
||||
for(int index = 0; index < printers.length(); index++)
|
||||
{
|
||||
JSONObject printer = printers.getJSONObject(index);
|
||||
@@ -482,15 +505,7 @@ public class PrintingService
|
||||
***************************************************************************/
|
||||
public JSONObject submit(JSONObject request) throws SQLException
|
||||
{
|
||||
String deviceId = required(request.optString("deviceId"), "deviceId");
|
||||
String printerId = required(request.optString("printerId"), "printerId");
|
||||
String key = required(request.optString("idempotencyKey"), "idempotencyKey");
|
||||
String title = required(request.optString("title"), "title");
|
||||
String format = required(request.optString("format"), "format");
|
||||
if(!Set.of("PDF", "ZPL").contains(format))
|
||||
{
|
||||
throw new PrintingException(400, "Only PDF and ZPL are supported");
|
||||
}
|
||||
byte[] content;
|
||||
try
|
||||
{
|
||||
@@ -512,47 +527,114 @@ public class PrintingService
|
||||
{
|
||||
throw new PrintingException(400, "PDF header is missing");
|
||||
}
|
||||
return (submitPrepared(request, hash(content), content.length, connection -> content));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Delay stored-document reads until atomic admission has succeeded.
|
||||
***************************************************************************/
|
||||
@FunctionalInterface
|
||||
private interface DocumentLoader
|
||||
{
|
||||
/***********************************************************************
|
||||
** Read the immutable document after capacity is reserved.
|
||||
***********************************************************************/
|
||||
byte[] load(Connection connection) throws SQLException;
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** All producers share durable admission before object storage allocation.
|
||||
***************************************************************************/
|
||||
private JSONObject submitPrepared(JSONObject request, String digest, int contentLength, DocumentLoader loader) throws SQLException
|
||||
{
|
||||
String deviceId = required(request.optString("deviceId"), "deviceId");
|
||||
String printerId = required(request.optString("printerId"), "printerId");
|
||||
String key = required(request.optString("idempotencyKey"), "idempotencyKey");
|
||||
String title = required(request.optString("title"), "title");
|
||||
String format = required(request.optString("format"), "format");
|
||||
if(!Set.of("PDF", "ZPL").contains(format))
|
||||
{
|
||||
throw new PrintingException(400, "Only PDF and ZPL are supported");
|
||||
}
|
||||
JSONObject options = printOptions(request);
|
||||
String digest = hash(content);
|
||||
String id = UUID.randomUUID().toString();
|
||||
String reference = storageTableName == null ? null : documentStorage.prefix() + id + "." + format.toLowerCase(java.util.Locale.ROOT);
|
||||
String binding = storageTableName == null ? null : documentStorage.binding(storageTableName);
|
||||
String uploadedId = null;
|
||||
try(Connection connection = dataSource.getConnection())
|
||||
{
|
||||
requireDevice(connection, deviceId);
|
||||
PrintingAdmission.initializeTenant(connection, tenantId);
|
||||
connection.setAutoCommit(false);
|
||||
try
|
||||
{
|
||||
long pendingDeadline = System.nanoTime() + java.util.concurrent.TimeUnit.SECONDS.toNanos(30);
|
||||
while(true)
|
||||
{
|
||||
PrintingAdmission.lock(connection, tenantId);
|
||||
requireDevice(connection, deviceId);
|
||||
query(connection, "SELECT id FROM printing_device WHERE id=? FOR UPDATE", deviceId);
|
||||
List<JSONObject> previous = query(connection, "SELECT " + JOB_COLUMNS + " FROM printing_job WHERE tenant_id=? AND idempotency_key=?", tenantId, key);
|
||||
if(!previous.isEmpty())
|
||||
{
|
||||
JSONObject old = previous.getFirst();
|
||||
if(!old.getString("device_id").equals(deviceId) || !old.getString("printer_id").equals(printerId) || !old.getString("content_hash").equals(digest) || !old.getString("format").equals(format) || !new JSONObject(old.getString("options_json")).similar(options))
|
||||
{
|
||||
throw new PrintingException(409, "Idempotency key was already used for a different request");
|
||||
}
|
||||
connection.commit();
|
||||
return (job(old));
|
||||
}
|
||||
if(query(connection, "SELECT id FROM printing_document_orphan WHERE tenant_id=? AND idempotency_key=? LIMIT 1", tenantId, key).isEmpty())
|
||||
{
|
||||
break;
|
||||
}
|
||||
connection.commit();
|
||||
if(System.nanoTime() >= pendingDeadline)
|
||||
{
|
||||
throw new PrintingException(429, "This request already has an upload in progress or awaiting recovery; retry later using the same request identity");
|
||||
}
|
||||
try
|
||||
{
|
||||
Thread.sleep(25);
|
||||
}
|
||||
catch(InterruptedException exception)
|
||||
{
|
||||
Thread.currentThread().interrupt();
|
||||
throw new SQLException("Interrupted while waiting for the existing submission", exception);
|
||||
}
|
||||
}
|
||||
List<JSONObject> printers = query(connection, "SELECT * FROM printing_printer WHERE device_id=? AND printer_id=? AND enabled=1 AND available=1", deviceId, printerId);
|
||||
if(printers.isEmpty() || !new JSONArray(printers.getFirst().getString("formats")).toList().contains(format))
|
||||
{
|
||||
throw new PrintingException(400, "Printer is not enabled for this format");
|
||||
}
|
||||
PrintingAdmission.reserve(connection, tenantId, deviceId, id, contentLength, admissionLimits);
|
||||
if(storageTableName != null)
|
||||
{
|
||||
requireDevice(connection, deviceId);
|
||||
reserveUpload(connection, id, storageTableName, reference, binding);
|
||||
reserveUpload(connection, id, storageTableName, reference, binding, deviceId, contentLength, key);
|
||||
connection.commit();
|
||||
uploadedId = id;
|
||||
if(query(connection, "SELECT id FROM printing_document_orphan WHERE id=? FOR UPDATE", id).isEmpty())
|
||||
{
|
||||
throw new PrintingException(409, "Upload reservation expired; retry submission");
|
||||
}
|
||||
}
|
||||
// Serialize submissions per device; the unique key also protects cross-device races.
|
||||
requireDevice(connection, deviceId);
|
||||
query(connection, "SELECT id FROM printing_device WHERE id=? FOR UPDATE", deviceId);
|
||||
List<JSONObject> previous = query(connection, "SELECT " + JOB_COLUMNS + " FROM printing_job WHERE tenant_id=? AND idempotency_key=?", tenantId, key);
|
||||
if(!previous.isEmpty())
|
||||
{
|
||||
JSONObject old = previous.getFirst();
|
||||
if(!old.getString("device_id").equals(deviceId) || !old.getString("printer_id").equals(printerId) || !old.getString("content_hash").equals(digest) || !old.getString("format").equals(format) || !new JSONObject(old.getString("options_json")).similar(options))
|
||||
requireDevice(connection, deviceId);
|
||||
query(connection, "SELECT id FROM printing_device WHERE id=? FOR UPDATE", deviceId);
|
||||
printers = query(connection, "SELECT * FROM printing_printer WHERE device_id=? AND printer_id=? AND enabled=1 AND available=1", deviceId, printerId);
|
||||
if(printers.isEmpty() || !new JSONArray(printers.getFirst().getString("formats")).toList().contains(format))
|
||||
{
|
||||
throw new PrintingException(409, "Idempotency key was already used for a different request");
|
||||
throw new PrintingException(400, "Printer is not enabled for this format");
|
||||
}
|
||||
update(connection, "DELETE FROM printing_document_orphan WHERE id=? AND tenant_id=?", id, tenantId);
|
||||
connection.commit();
|
||||
return (job(old));
|
||||
}
|
||||
List<JSONObject> printers = query(connection, "SELECT * FROM printing_printer WHERE device_id=? AND printer_id=? AND enabled=1 AND available=1", deviceId, printerId);
|
||||
if(printers.isEmpty() || !new JSONArray(printers.getFirst().getString("formats")).toList().contains(format))
|
||||
byte[] content = loader.load(connection);
|
||||
if(content.length != contentLength || !hash(content).equals(digest))
|
||||
{
|
||||
throw new PrintingException(400, "Printer is not enabled for this format");
|
||||
throw new PrintingException(409, "Stored print document integrity validation failed");
|
||||
}
|
||||
if(storageTableName != null)
|
||||
{
|
||||
@@ -676,7 +758,8 @@ public class PrintingService
|
||||
try(Connection connection = dataSource.getConnection())
|
||||
{
|
||||
requireDevice(connection, deviceId);
|
||||
row = deviceJob(connection, deviceId, jobId, true);
|
||||
row = deviceJob(connection, deviceId, jobId);
|
||||
row.put("content_available", query(connection, "SELECT content_available FROM printing_job WHERE id=? AND tenant_id=? AND device_id=?", jobId, tenantId, deviceId).getFirst().getInt("content_available"));
|
||||
List<JSONObject> previous = query(connection, "SELECT " + JOB_COLUMNS + " FROM printing_job WHERE tenant_id=? AND device_id=? AND idempotency_key=?", tenantId, deviceId, retryKey);
|
||||
if(!previous.isEmpty())
|
||||
{
|
||||
@@ -698,10 +781,10 @@ public class PrintingService
|
||||
{
|
||||
throw new PrintingException(409, "Inspect the printer and confirm the risk of duplicate output before reprinting");
|
||||
}
|
||||
return (submit(new JSONObject().put("deviceId", deviceId).put("printerId", row.getString("printer_id"))
|
||||
.put("title", row.getString("title")).put("format", row.getString("format")).put("contentBase64", Base64.getEncoder().encodeToString(documentBytes(row)))
|
||||
return (submitPrepared(new JSONObject().put("deviceId", deviceId).put("printerId", row.getString("printer_id"))
|
||||
.put("title", row.getString("title")).put("format", row.getString("format"))
|
||||
.put("options", new JSONObject(row.getString("options_json")))
|
||||
.put("idempotencyKey", retryKey)));
|
||||
.put("idempotencyKey", retryKey), row.getString("content_hash"), row.getInt("content_length"), connection -> documentBytes(deviceJob(connection, deviceId, jobId, true))));
|
||||
}
|
||||
|
||||
|
||||
@@ -884,25 +967,68 @@ public class PrintingService
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Return printers for a device owned by this tenant.
|
||||
** Bounded inventory history for management clients.
|
||||
***************************************************************************/
|
||||
public JSONArray printers(String deviceId) throws SQLException
|
||||
public JSONObject printersPage(String deviceId, int limit, String before) throws SQLException
|
||||
{
|
||||
if(limit < 1 || limit > 200)
|
||||
{
|
||||
throw new PrintingException(400, "Printer page limit must be between 1 and 200");
|
||||
}
|
||||
String name = null;
|
||||
String id = null;
|
||||
if(before != null && !before.isEmpty())
|
||||
{
|
||||
try
|
||||
{
|
||||
if(before.length() > 12000)
|
||||
{
|
||||
throw new IllegalArgumentException("Cursor too long");
|
||||
}
|
||||
JSONObject cursor = new JSONObject(new String(Base64.getUrlDecoder().decode(before), StandardCharsets.UTF_8));
|
||||
name = optionalText(cursor.getString("name"), "cursor name", 1024);
|
||||
id = UUID.fromString(cursor.getString("id")).toString();
|
||||
}
|
||||
catch(RuntimeException exception)
|
||||
{
|
||||
throw new PrintingException(400, "Invalid printer cursor");
|
||||
}
|
||||
}
|
||||
try(Connection connection = dataSource.getConnection())
|
||||
{
|
||||
requireDevice(connection, deviceId);
|
||||
String sql = "SELECT * FROM printing_printer WHERE tenant_id=? AND device_id=?";
|
||||
List<JSONObject> rows = name == null ? query(connection, sql + " ORDER BY name,id LIMIT ?", tenantId, deviceId, limit + 1)
|
||||
: query(connection, sql + " AND (name>? OR (name=? AND id>?)) ORDER BY name,id LIMIT ?", tenantId, deviceId, name, name, id, limit + 1);
|
||||
JSONArray result = new JSONArray();
|
||||
for(JSONObject row : query(connection, "SELECT * FROM printing_printer WHERE device_id=? ORDER BY name", deviceId))
|
||||
for(JSONObject row : rows.subList(0, Math.min(limit, rows.size())))
|
||||
{
|
||||
result.put(new JSONObject().put("printerId", row.getString("printer_id")).put("name", row.getString("queue_name")).put("displayName", row.optString("display_name")).put("description", row.optString("description")).put("location", row.optString("location"))
|
||||
.put("driverName", row.optString("driver_name")).put("portName", row.optString("port_name")).put("status", row.getString("status")).put("formats", new JSONArray(row.getString("formats"))).put("enabled", row.getInt("enabled") == 1).put("available", row.getInt("available") == 1).put("lastSeenAt", row.getString("last_seen_at")));
|
||||
}
|
||||
return (result);
|
||||
JSONObject response = new JSONObject().put("printers", result).put("nextCursor", JSONObject.NULL);
|
||||
if(rows.size() > limit)
|
||||
{
|
||||
JSONObject last = rows.get(limit - 1);
|
||||
String cursor = new JSONObject().put("name", last.getString("name")).put("id", last.getString("id")).toString();
|
||||
response.put("nextCursor", Base64.getUrlEncoder().withoutPadding().encodeToString(cursor.getBytes(StandardCharsets.UTF_8)));
|
||||
}
|
||||
return (response);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Existing callers retain their array contract with a bounded first page.
|
||||
***************************************************************************/
|
||||
public JSONArray printers(String deviceId) throws SQLException
|
||||
{
|
||||
return (printersPage(deviceId, 200, null).getJSONArray("printers"));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Return recent jobs; content is excluded from management lists.
|
||||
***************************************************************************/
|
||||
@@ -1025,6 +1151,7 @@ public class PrintingService
|
||||
}
|
||||
String cutoff = Instant.now().minusSeconds(policy.days() * 86400L).toString();
|
||||
{
|
||||
PrintingAdmission.initializeTenant(connection, tenantId);
|
||||
connection.setAutoCommit(false);
|
||||
try
|
||||
{
|
||||
@@ -1033,7 +1160,8 @@ public class PrintingService
|
||||
for(JSONObject row : rows)
|
||||
{
|
||||
renewRetentionLease(connection, leaseToken);
|
||||
List<JSONObject> locked = query(connection, "SELECT id, format, content_storage_table, content_reference, content_binding FROM printing_job WHERE tenant_id=? AND id=? AND content_available=1 AND status IN ('SPOOLED','SPOOLER_COMPLETED') AND completed_at<? FOR UPDATE", tenantId, row.getString("id"), cutoff);
|
||||
PrintingAdmission.lock(connection, tenantId);
|
||||
List<JSONObject> locked = query(connection, "SELECT id, device_id, content_length, format, content_storage_table, content_reference, content_binding FROM printing_job WHERE tenant_id=? AND id=? AND content_available=1 AND status IN ('SPOOLED','SPOOLER_COMPLETED') AND completed_at<? FOR UPDATE", tenantId, row.getString("id"), cutoff);
|
||||
if(locked.isEmpty())
|
||||
{
|
||||
continue;
|
||||
@@ -1045,6 +1173,10 @@ public class PrintingService
|
||||
documentStorage.delete(document.getString("content_storage_table"), document.getString("content_reference"));
|
||||
}
|
||||
int changed = update(connection, "UPDATE printing_job SET content_base64=NULL, content_available=0 WHERE tenant_id=? AND id=? AND content_available=1 AND status IN ('SPOOLED','SPOOLER_COMPLETED') AND completed_at<?", tenantId, row.getString("id"), cutoff);
|
||||
if(changed == 1)
|
||||
{
|
||||
PrintingAdmission.release(connection, tenantId, document.getString("device_id"), document.getLong("content_length"));
|
||||
}
|
||||
connection.commit();
|
||||
purged += changed;
|
||||
progress[0] += changed;
|
||||
@@ -1214,6 +1346,10 @@ public class PrintingService
|
||||
***************************************************************************/
|
||||
private byte[] documentBytes(JSONObject row) throws SQLException
|
||||
{
|
||||
if(row.getInt("content_available") == 0)
|
||||
{
|
||||
throw new PrintingException(410, "Print document expired under the host retention policy; submit a new document explicitly");
|
||||
}
|
||||
byte[] bytes;
|
||||
if(row.isNull("content_reference"))
|
||||
{
|
||||
@@ -1255,9 +1391,9 @@ public class PrintingService
|
||||
/***************************************************************************
|
||||
** A committed upload intent survives crashes between object and SQL writes.
|
||||
***************************************************************************/
|
||||
private void reserveUpload(Connection connection, String id, String table, String reference, String binding) throws SQLException
|
||||
private void reserveUpload(Connection connection, String id, String table, String reference, String binding, String deviceId, int contentLength, String key) throws SQLException
|
||||
{
|
||||
update(connection, "INSERT INTO printing_document_orphan (id, tenant_id, storage_table, reference, binding, created_at) VALUES (?, ?, ?, ?, ?, ?)", id, tenantId, table, reference, binding, Instant.now().toString());
|
||||
update(connection, "INSERT INTO printing_document_orphan (id, tenant_id, storage_table, reference, binding, created_at, device_id, content_length, idempotency_key) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", id, tenantId, table, reference, binding, Instant.now().toString(), deviceId, contentLength, key);
|
||||
}
|
||||
|
||||
|
||||
@@ -1272,6 +1408,7 @@ public class PrintingService
|
||||
connection.setAutoCommit(false);
|
||||
try
|
||||
{
|
||||
PrintingAdmission.lock(connection, tenantId);
|
||||
List<JSONObject> rows = query(connection, "SELECT * FROM printing_document_orphan WHERE tenant_id=? AND id=? FOR UPDATE", tenantId, id);
|
||||
if(rows.isEmpty())
|
||||
{
|
||||
@@ -1293,6 +1430,7 @@ public class PrintingService
|
||||
throw new PrintingException(409, "Orphan reference does not match its owned upload identity");
|
||||
}
|
||||
documentStorage.delete(orphan.getString("storage_table"), orphan.getString("reference"));
|
||||
PrintingAdmission.release(connection, tenantId, orphan.isNull("device_id") ? null : orphan.getString("device_id"), orphan.isNull("content_length") ? PrintingAdmission.UNKNOWN_UPLOAD_BYTES : orphan.getLong("content_length"));
|
||||
}
|
||||
update(connection, "DELETE FROM printing_document_orphan WHERE id=? AND tenant_id=?", id, tenantId);
|
||||
connection.commit();
|
||||
|
||||
@@ -0,0 +1,601 @@
|
||||
/*
|
||||
* 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.sql.SQLException;
|
||||
import java.time.Instant;
|
||||
import java.util.Base64;
|
||||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.Callable;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
|
||||
import org.json.JSONArray;
|
||||
import org.json.JSONObject;
|
||||
import org.junit.jupiter.api.Test;
|
||||
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;
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Security regressions use isolated SQL and synthetic object storage only.
|
||||
*******************************************************************************/
|
||||
class PrintingAdmissionTest
|
||||
{
|
||||
private static final byte[] DOCUMENT = new byte[1024];
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Tiny retry requests cannot accumulate unlimited retained terminal copies.
|
||||
***************************************************************************/
|
||||
@Test
|
||||
void retainedRetryBudgetRejectsBeforeUploadAndPreservesReplayAfterCleanup() throws Exception
|
||||
{
|
||||
Fixture fixture = fixture(limits(10, 20, 100_000, 200_000, 2048, 4096, 100, 200, 10, 20));
|
||||
String original = fixture.service.submit(request(fixture.first, "original")).getString("jobId");
|
||||
spool(fixture.service, fixture.first, original);
|
||||
JSONObject retryRequest = retry("same-retry");
|
||||
String copy = fixture.service.retry(fixture.first, original, retryRequest).getString("jobId");
|
||||
spool(fixture.service, fixture.first, copy);
|
||||
int readsBeforeRejection = fixture.documents.gets.get();
|
||||
assertQuota(() -> fixture.service.retry(fixture.first, original, retry("fresh-retry")));
|
||||
assertEquals(readsBeforeRejection, fixture.documents.gets.get());
|
||||
assertEquals(copy, fixture.service.retry(fixture.first, original, retryRequest).getString("jobId"));
|
||||
assertEquals(2, fixture.documents.puts.get());
|
||||
assertEquals(2, count(fixture.source, "printing_job"));
|
||||
execute(fixture.source, "UPDATE printing_job SET completed_at=? WHERE id=?", Instant.now().minusSeconds(400L * 86400).toString(), original);
|
||||
assertEquals(1, fixture.service.runRetention());
|
||||
assertEquals(copy, fixture.service.retry(fixture.first, original, retryRequest).getString("jobId"));
|
||||
assertEquals(410, assertThrows(PrintingException.class, () -> fixture.service.retry(fixture.first, original, retry("expired-source"))).getStatus());
|
||||
fixture.service.submit(request(fixture.first, "after-cleanup"));
|
||||
assertEquals(3, fixture.documents.puts.get());
|
||||
assertEquals(2, fixture.documents.objects.size());
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Tenant count and byte budgets serialize across service instances/devices.
|
||||
***************************************************************************/
|
||||
@Test
|
||||
void concurrentDevicesCannotExceedTenantCountOrByteBudgets() throws Exception
|
||||
{
|
||||
List<PrintingAdmissionLimits> budgets = List.of(
|
||||
limits(10, 1, 100_000, 200_000, 100_000, 200_000, 100, 200, 10, 20),
|
||||
limits(10, 20, 100_000, DOCUMENT.length, 100_000, 200_000, 100, 200, 10, 20),
|
||||
limits(10, 20, 100_000, 200_000, 100_000, DOCUMENT.length, 100, 200, 10, 20));
|
||||
for(int attempt = 0; attempt < 10; attempt++)
|
||||
{
|
||||
for(PrintingAdmissionLimits budget : budgets)
|
||||
{
|
||||
Fixture fixture = fixture(budget);
|
||||
PrintingService other = sibling(fixture, budget);
|
||||
concurrentOneAccepted(() -> fixture.service.submit(request(fixture.first, "first")), () -> other.submit(request(fixture.second, "second")));
|
||||
assertEquals(1, fixture.documents.puts.get());
|
||||
assertEquals(1, count(fixture.source, "printing_job"));
|
||||
assertEquals(0, count(fixture.source, "printing_document_orphan"));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Device budgets remain effective even when tenant capacity is available.
|
||||
***************************************************************************/
|
||||
@Test
|
||||
void outstandingDeviceCountAndBytesReleaseOnlyWhenWorkFinishes() throws Exception
|
||||
{
|
||||
for(PrintingAdmissionLimits budget : List.of(
|
||||
limits(1, 20, 100_000, 200_000, 100_000, 200_000, 100, 200, 10, 20),
|
||||
limits(10, 20, DOCUMENT.length, 200_000, 100_000, 200_000, 100, 200, 10, 20)))
|
||||
{
|
||||
Fixture fixture = fixture(budget);
|
||||
String original = fixture.service.submit(request(fixture.first, "original")).getString("jobId");
|
||||
assertQuota(() -> fixture.service.submit(request(fixture.first, "overflow")));
|
||||
assertEquals(original, fixture.service.submit(request(fixture.first, "original")).getString("jobId"));
|
||||
fixture.service.submit(request(fixture.second, "other-device"));
|
||||
assertEquals(2, fixture.documents.puts.get());
|
||||
spool(fixture.service, fixture.first, original);
|
||||
fixture.service.submit(request(fixture.first, "finished-slot"));
|
||||
assertEquals(3, fixture.documents.puts.get());
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Failed upload compensation keeps its full reservation until exact cleanup.
|
||||
***************************************************************************/
|
||||
@Test
|
||||
void failedUploadIntentConsumesCapacityUntilCleanupSucceeds() throws Exception
|
||||
{
|
||||
Fixture fixture = fixture(limits(10, 20, 100_000, 200_000, DOCUMENT.length, DOCUMENT.length, 100, 200, 10, 20));
|
||||
fixture.documents.failUpload = true;
|
||||
fixture.documents.failDelete = true;
|
||||
assertThrows(SQLException.class, () -> fixture.service.submit(request(fixture.first, "failed-upload")));
|
||||
assertEquals(1, count(fixture.source, "printing_document_orphan"));
|
||||
assertEquals(0, count(fixture.source, "printing_job"));
|
||||
assertQuota(() -> fixture.service.submit(request(fixture.second, "blocked-by-reservation")));
|
||||
assertEquals(1, fixture.documents.puts.get());
|
||||
fixture.documents.failUpload = false;
|
||||
fixture.documents.failDelete = false;
|
||||
execute(fixture.source, "UPDATE printing_document_orphan SET created_at=?", Instant.now().minusSeconds(7200).toString());
|
||||
assertEquals(1, fixture.service.runRetention());
|
||||
fixture.service.submit(request(fixture.second, "after-recovery"));
|
||||
assertEquals(2, fixture.documents.puts.get());
|
||||
assertEquals(1, fixture.documents.objects.size());
|
||||
assertEquals(0, count(fixture.source, "printing_document_orphan"));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Legacy upload intents with unknown size reserve the maximum allowed bytes.
|
||||
***************************************************************************/
|
||||
@Test
|
||||
void legacyUnknownUploadChargesTenantCapacityUntilExactCleanup() throws Exception
|
||||
{
|
||||
long unknownBytes = 20L * 1024 * 1024;
|
||||
Fixture fixture = fixture(limits(10, 20, 100_000_000, 200_000_000, 100_000_000, unknownBytes, 100, 200, 10, 20));
|
||||
String orphan = UUID.randomUUID().toString();
|
||||
String reference = fixture.documents.prefix() + orphan + ".zpl";
|
||||
fixture.documents.objects.put(reference, DOCUMENT.clone());
|
||||
execute(fixture.source, "INSERT INTO printing_document_orphan (id,tenant_id,storage_table,reference,binding,created_at) VALUES (?,?,?,?,?,?)",
|
||||
orphan, "tenant-a", "printing_documents", reference, "quota-fixture", Instant.now().minusSeconds(7200).toString());
|
||||
// Only isolated test accounting tables are removed to exercise upgrade backfill.
|
||||
execute(fixture.source, "DROP TABLE printing_resource_usage");
|
||||
execute(fixture.source, "DROP TABLE printing_admission_event");
|
||||
fixture.service.initialize();
|
||||
assertEquals(Long.toString(unknownBytes), scalar(fixture.source, "SELECT retained_bytes FROM printing_resource_usage WHERE tenant_id=? AND device_id=''", "tenant-a"));
|
||||
assertEquals("0", scalar(fixture.source, "SELECT retained_bytes FROM printing_resource_usage WHERE tenant_id=? AND device_id=?", "tenant-a", fixture.first));
|
||||
assertQuota(() -> fixture.service.submit(request(fixture.first, "blocked-by-legacy-upload")));
|
||||
assertEquals(0, fixture.documents.puts.get());
|
||||
assertEquals(0, fixture.documents.gets.get());
|
||||
assertEquals(1, fixture.documents.objects.size());
|
||||
assertTrue(fixture.documents.objects.containsKey(reference));
|
||||
assertEquals(1, count(fixture.source, "printing_document_orphan"));
|
||||
assertEquals(0, count(fixture.source, "printing_job"));
|
||||
assertEquals(1, fixture.service.runRetention());
|
||||
assertFalse(fixture.documents.objects.containsKey(reference));
|
||||
assertEquals("0", scalar(fixture.source, "SELECT retained_bytes FROM printing_resource_usage WHERE tenant_id=? AND device_id=''", "tenant-a"));
|
||||
fixture.service.submit(request(fixture.second, "after-legacy-cleanup"));
|
||||
assertEquals(1, fixture.documents.puts.get());
|
||||
assertEquals(1, fixture.documents.objects.size());
|
||||
assertEquals(0, count(fixture.source, "printing_document_orphan"));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Finished jobs do not permit unlimited small-request amplification per hour.
|
||||
***************************************************************************/
|
||||
@Test
|
||||
void hourlyCreationBudgetIncludesFinishedRetriesAndDirectSubmission() throws Exception
|
||||
{
|
||||
Fixture fixture = fixture(limits(10, 20, 100_000, 200_000, 100_000, 200_000, 2, 100, 10, 20));
|
||||
String original = fixture.service.submit(request(fixture.first, "original")).getString("jobId");
|
||||
spool(fixture.service, fixture.first, original);
|
||||
JSONObject same = retry("first-copy");
|
||||
String copy = fixture.service.retry(fixture.first, original, same).getString("jobId");
|
||||
spool(fixture.service, fixture.first, copy);
|
||||
int readsBeforeRejection = fixture.documents.gets.get();
|
||||
assertQuota(() -> fixture.service.retry(fixture.first, original, retry("second-copy")));
|
||||
assertEquals(readsBeforeRejection, fixture.documents.gets.get());
|
||||
assertQuota(() -> fixture.service.submit(request(fixture.first, "direct-bypass")));
|
||||
assertEquals(copy, fixture.service.retry(fixture.first, original, same).getString("jobId"));
|
||||
assertEquals(2, fixture.documents.puts.get());
|
||||
fixture.service.submit(request(fixture.second, "independent-device"));
|
||||
assertEquals(3, fixture.documents.puts.get());
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Rotated IDs cannot erase history or bypass cumulative device capacity.
|
||||
***************************************************************************/
|
||||
@Test
|
||||
void inventoryRotationRejectsAtomicallyAndPreservesApprovalAndJobReference() throws Exception
|
||||
{
|
||||
Fixture fixture = fixture(limits(10, 20, 100_000, 200_000, 100_000, 200_000, 100, 200, 3, 20));
|
||||
fixture.service.inventory(fixture.first, new JSONArray().put(printer("queue").put("displayName", "Approved queue").put("description", "Historical description")));
|
||||
String job = fixture.service.submit(request(fixture.first, "history")).getString("jobId");
|
||||
String rowId = scalar(fixture.source, "SELECT printer_row_id FROM printing_job WHERE id=?", job);
|
||||
fixture.service.inventory(fixture.first, new JSONArray().put(printer("new-a")).put(printer("new-b")));
|
||||
assertQuota(() ->
|
||||
{
|
||||
fixture.service.inventory(fixture.first, new JSONArray().put(printer("overflow")));
|
||||
return null;
|
||||
});
|
||||
assertEquals("1", scalar(fixture.source, "SELECT available FROM printing_printer WHERE device_id=? AND printer_id=?", fixture.first, "new-a"));
|
||||
assertEquals("1", scalar(fixture.source, "SELECT enabled FROM printing_printer WHERE id=?", rowId));
|
||||
assertEquals("Historical description", scalar(fixture.source, "SELECT description FROM printing_printer WHERE id=?", rowId));
|
||||
assertEquals(rowId, scalar(fixture.source, "SELECT printer_row_id FROM printing_job WHERE id=?", job));
|
||||
fixture.service.inventory(fixture.first, new JSONArray().put(printer("queue").put("displayName", "Updated approved queue")));
|
||||
assertEquals("Updated approved queue", scalar(fixture.source, "SELECT name FROM printing_printer WHERE id=?", rowId));
|
||||
assertEquals("1", scalar(fixture.source, "SELECT enabled FROM printing_printer WHERE id=?", rowId));
|
||||
assertEquals(4, count(fixture.source, "printing_printer"));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Two devices cannot simultaneously claim the last tenant inventory slot.
|
||||
***************************************************************************/
|
||||
@Test
|
||||
void concurrentInventoriesEnforceTenantCapacityAcrossServiceInstances() throws Exception
|
||||
{
|
||||
PrintingAdmissionLimits budget = limits(10, 20, 100_000, 200_000, 100_000, 200_000, 100, 200, 10, 3);
|
||||
Fixture fixture = fixture(budget);
|
||||
PrintingService other = sibling(fixture, budget);
|
||||
concurrentOneAccepted(() ->
|
||||
{
|
||||
fixture.service.inventory(fixture.first, new JSONArray().put(printer("new-first")));
|
||||
return null;
|
||||
}, () ->
|
||||
{
|
||||
other.inventory(fixture.second, new JSONArray().put(printer("new-second")));
|
||||
return null;
|
||||
});
|
||||
assertEquals(3, count(fixture.source, "printing_printer"));
|
||||
assertEquals("2", scalar(fixture.source, "SELECT COUNT(*) FROM printing_printer WHERE enabled=1"));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Legacy over-budget history remains accessible through bounded pages.
|
||||
***************************************************************************/
|
||||
@Test
|
||||
void existingOversizedInventoryIsPagedWithoutDeletingHistoricalRows() throws Exception
|
||||
{
|
||||
Fixture fixture = fixture(limits(10, 20, 100_000, 200_000, 100_000, 200_000, 100, 200, 1000, 2000));
|
||||
for(int batch = 0; batch < 2; batch++)
|
||||
{
|
||||
JSONArray printers = new JSONArray();
|
||||
for(int index = 0; index < 150; index++)
|
||||
{
|
||||
printers.put(printer("historic-" + (batch * 150 + index)).put("name", "倉庫".repeat(510) + String.format(java.util.Locale.ROOT, "%04d", batch * 150 + index)));
|
||||
}
|
||||
fixture.service.inventory(fixture.first, printers);
|
||||
}
|
||||
fixture.service.withAdmissionLimits(limits(10, 20, 100_000, 200_000, 100_000, 200_000, 100, 200, 3, 5));
|
||||
assertEquals(200, fixture.service.printers(fixture.first).length());
|
||||
Set<String> ids = new HashSet<>();
|
||||
String cursor = null;
|
||||
int pages = 0;
|
||||
do
|
||||
{
|
||||
JSONObject page = fixture.service.printersPage(fixture.first, 73, cursor);
|
||||
JSONArray rows = page.getJSONArray("printers");
|
||||
assertTrue(rows.length() <= 73);
|
||||
for(int index = 0; index < rows.length(); index++)
|
||||
{
|
||||
assertTrue(ids.add(rows.getJSONObject(index).getString("printerId")));
|
||||
}
|
||||
cursor = page.optString("nextCursor", "");
|
||||
assertTrue(++pages <= 10);
|
||||
}
|
||||
while(!cursor.isBlank());
|
||||
assertEquals(301, ids.size());
|
||||
assertEquals("0", scalar(fixture.source, "SELECT COUNT(*) FROM INFORMATION_SCHEMA.INDEX_COLUMNS WHERE UPPER(TABLE_NAME)='PRINTING_PRINTER' AND UPPER(COLUMN_NAME)='NAME'"));
|
||||
fixture.service.inventory(fixture.first, new JSONArray().put(printer("queue")));
|
||||
assertEquals(302, count(fixture.source, "printing_printer"));
|
||||
assertQuota(() ->
|
||||
{
|
||||
fixture.service.inventory(fixture.first, new JSONArray().put(printer("extra")));
|
||||
return null;
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Exercise two independent service objects against one tenant database.
|
||||
***************************************************************************/
|
||||
private static PrintingService sibling(Fixture fixture, PrintingAdmissionLimits limits)
|
||||
{
|
||||
PrintingService service = new PrintingService(fixture.source, "tenant-a").withAdmissionLimits(limits);
|
||||
service.configureStorage(fixture.documents, "printing_documents");
|
||||
return (service);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Synthetic storage and isolated SQL never reach an installed print agent.
|
||||
***************************************************************************/
|
||||
private static Fixture fixture(PrintingAdmissionLimits limits) throws Exception
|
||||
{
|
||||
DriverDataSource source = new DriverDataSource("jdbc:h2:mem:admission-" + UUID.randomUUID() + ";DB_CLOSE_DELAY=-1;LOCK_TIMEOUT=5000", "sa", "");
|
||||
PrintingService service = new PrintingService(source, "tenant-a").withAdmissionLimits(limits);
|
||||
service.initialize();
|
||||
CountingDocuments documents = new CountingDocuments();
|
||||
service.configureStorage(documents, "printing_documents");
|
||||
return (new Fixture(source, service, documents, enroll(service, "first"), enroll(service, "second")));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Every printer approval remains explicit in the fixture.
|
||||
***************************************************************************/
|
||||
private static String enroll(PrintingService service, String name) throws Exception
|
||||
{
|
||||
String device = service.enroll(service.createEnrollment("warehouse", name).put("deviceName", name).put("agentVersion", "test")).getString("deviceId");
|
||||
service.inventory(device, new JSONArray().put(printer("queue")));
|
||||
service.enablePrinter(device, "queue", true);
|
||||
return (device);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** A current queue identity, distinct from its display metadata.
|
||||
***************************************************************************/
|
||||
private static JSONObject printer(String id)
|
||||
{
|
||||
return (new JSONObject().put("printerId", id).put("name", "Synthetic " + id).put("status", "READY").put("formats", new JSONArray(List.of("ZPL"))));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Exact synthetic bytes make byte budgets independent of transport encoding.
|
||||
***************************************************************************/
|
||||
private static JSONObject request(String device, String key)
|
||||
{
|
||||
return (new JSONObject().put("deviceId", device).put("printerId", "queue").put("title", "Admission regression").put("format", "ZPL")
|
||||
.put("contentBase64", Base64.getEncoder().encodeToString(DOCUMENT)).put("idempotencyKey", key));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Reprints retain explicit operator duplicate-risk consent.
|
||||
***************************************************************************/
|
||||
private static JSONObject retry(String key)
|
||||
{
|
||||
return (new JSONObject().put("requestId", key).put("confirmDuplicateRisk", true));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Use real event transitions so accounting sees the same production path.
|
||||
***************************************************************************/
|
||||
private static void spool(PrintingService service, String device, String job) throws Exception
|
||||
{
|
||||
assertEquals(job, service.claim(device).getJSONObject("job").getString("jobId"));
|
||||
service.event(device, job, new JSONObject().put("eventId", UUID.randomUUID().toString()).put("status", "SPOOLED").put("spoolerJobId", "synthetic"));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Over-budget requests must fail as operator-visible backpressure.
|
||||
***************************************************************************/
|
||||
private static void assertQuota(Callable<?> operation)
|
||||
{
|
||||
PrintingException exception = assertThrows(PrintingException.class, operation::call);
|
||||
assertEquals(429, exception.getStatus());
|
||||
assertFalse(exception.getMessage().isBlank());
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Start both calls together; only one may consume the final shared slot.
|
||||
***************************************************************************/
|
||||
private static void concurrentOneAccepted(Callable<?> first, Callable<?> second) throws Exception
|
||||
{
|
||||
CountDownLatch start = new CountDownLatch(1);
|
||||
try(var pool = Executors.newFixedThreadPool(2))
|
||||
{
|
||||
var a = pool.submit(() -> outcome(start, first));
|
||||
var b = pool.submit(() -> outcome(start, second));
|
||||
start.countDown();
|
||||
assertEquals(List.of("accepted", "quota"), List.of(a.get(10, TimeUnit.SECONDS), b.get(10, TimeUnit.SECONDS)).stream().sorted().toList());
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Unexpected exceptions remain test failures rather than rejected admissions.
|
||||
***************************************************************************/
|
||||
private static String outcome(CountDownLatch start, Callable<?> operation) throws Exception
|
||||
{
|
||||
start.await();
|
||||
try
|
||||
{
|
||||
operation.call();
|
||||
return ("accepted");
|
||||
}
|
||||
catch(PrintingException exception)
|
||||
{
|
||||
assertEquals(429, exception.getStatus());
|
||||
return ("quota");
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Explicit arguments keep each boundary under test visible at its call site.
|
||||
***************************************************************************/
|
||||
private static PrintingAdmissionLimits limits(int deviceJobs, int tenantJobs, long deviceBytes, long tenantBytes, long deviceRetained,
|
||||
long tenantRetained, int deviceRate, int tenantRate, int devicePrinters, int tenantPrinters)
|
||||
{
|
||||
return (new PrintingAdmissionLimits(deviceJobs, tenantJobs, deviceBytes, tenantBytes, deviceRetained, tenantRetained, deviceRate, tenantRate, devicePrinters, tenantPrinters));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Count only fixed test-owned tables.
|
||||
***************************************************************************/
|
||||
private static int count(DriverDataSource source, String table) throws Exception
|
||||
{
|
||||
return (Integer.parseInt(scalar(source, "SELECT COUNT(*) FROM " + table)));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Parameterized inspection avoids document reads when checking receipts.
|
||||
***************************************************************************/
|
||||
private static String scalar(DriverDataSource source, String sql, Object... values) throws Exception
|
||||
{
|
||||
try(var connection = source.getConnection(); var statement = connection.prepareStatement(sql))
|
||||
{
|
||||
for(int index = 0; index < values.length; index++)
|
||||
{
|
||||
statement.setObject(index + 1, values[index]);
|
||||
}
|
||||
try(var rows = statement.executeQuery())
|
||||
{
|
||||
assertTrue(rows.next());
|
||||
return (rows.getString(1));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Age only isolated fixture rows to exercise retention recovery.
|
||||
***************************************************************************/
|
||||
private static void execute(DriverDataSource source, String sql, Object... values) throws Exception
|
||||
{
|
||||
try(var connection = source.getConnection(); var statement = connection.prepareStatement(sql))
|
||||
{
|
||||
for(int index = 0; index < values.length; index++)
|
||||
{
|
||||
statement.setObject(index + 1, values[index]);
|
||||
}
|
||||
statement.executeUpdate();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Isolated state shared only by one test and its two service instances.
|
||||
***************************************************************************/
|
||||
private record Fixture(DriverDataSource source, PrintingService service, CountingDocuments documents, String first, String second)
|
||||
{
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***************************************************************************
|
||||
** Count object writes and inject ambiguous storage failures without network.
|
||||
***************************************************************************/
|
||||
private static class CountingDocuments extends PrintingDocumentStorage
|
||||
{
|
||||
private final Map<String, byte[]> objects = new ConcurrentHashMap<>();
|
||||
private final AtomicInteger puts = new AtomicInteger();
|
||||
private final AtomicInteger gets = new AtomicInteger();
|
||||
private boolean failUpload;
|
||||
private boolean failDelete;
|
||||
|
||||
|
||||
|
||||
/***********************************************************************
|
||||
** Reference ownership uses the same tenant prefix as production.
|
||||
***********************************************************************/
|
||||
CountingDocuments()
|
||||
{
|
||||
super(new QInstance(), "tenant-a", true);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***********************************************************************
|
||||
** Metadata validation belongs to the separate native storage tests.
|
||||
***********************************************************************/
|
||||
@Override
|
||||
public String binding(String table)
|
||||
{
|
||||
return ("quota-fixture");
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***********************************************************************
|
||||
** A failing upload can still have created the entire remote object.
|
||||
***********************************************************************/
|
||||
@Override
|
||||
public void put(String table, String reference, String format, byte[] content) throws SQLException
|
||||
{
|
||||
puts.incrementAndGet();
|
||||
objects.put(reference, content.clone());
|
||||
if(failUpload)
|
||||
{
|
||||
throw new SQLException("Synthetic upload acknowledgement failure");
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***********************************************************************
|
||||
** Read immutable synthetic bytes for the retry path.
|
||||
***********************************************************************/
|
||||
@Override
|
||||
public byte[] get(String table, String reference) throws SQLException
|
||||
{
|
||||
gets.incrementAndGet();
|
||||
byte[] content = objects.get(reference);
|
||||
if(content == null)
|
||||
{
|
||||
throw new SQLException("Synthetic object was removed");
|
||||
}
|
||||
return (content.clone());
|
||||
}
|
||||
|
||||
|
||||
|
||||
/***********************************************************************
|
||||
** Failed cleanup must keep its reservation; successful deletion is exact.
|
||||
***********************************************************************/
|
||||
@Override
|
||||
public void delete(String table, String reference) throws SQLException
|
||||
{
|
||||
if(failDelete)
|
||||
{
|
||||
throw new SQLException("Synthetic delete failure");
|
||||
}
|
||||
objects.remove(reference);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user