diff --git a/archetype-validation-prober/PYTHON_COMPANION.md b/archetype-validation-prober/PYTHON_COMPANION.md new file mode 100644 index 00000000..82fc5c3f --- /dev/null +++ b/archetype-validation-prober/PYTHON_COMPANION.md @@ -0,0 +1,41 @@ +# Python Companion for the Archetype Gate + +This repository keeps the Java prober as the upstream reference, but the same +idea can be implemented in a more compact and operationally friendly way with a +Python companion. + +## What the companion adds + +- `Archetype` gate implemented as a total function in Python +- `ValidationResult` and `ValidationError` as immutable value objects +- `ClassifyingReceiver` without a builder or singleton boilerplate +- `vertex_agent/` for ingestion, retrieval, and MCP exposure +- real GCP resources: + - Pub/Sub topic and subscription + - BigQuery dataset and table + - Cloud Storage bucket for docs +- smoke validation with OAuth/ADC only + +## Why this is useful + +The Java prober proves the gate behavior. The Python companion shows how the +same architecture can be expanded into a full GCP-native workflow: + +1. validate deterministically before publish +2. keep the retry path for transient failures only +3. store knowledge in BigQuery +4. expose the result through MCP for coding agents +5. validate everything against real infrastructure + +## What was proven in practice + +- `uv run pytest` passed +- `uv run python -m scripts.gcp_smoke` returned `smoke=ok` +- Pub/Sub publish/pull round-trip succeeded +- BigQuery insert/query succeeded + +## Security note + +No service account keys are stored in the repository. Real smoke tests use +developer-owned OAuth/ADC credentials only. + diff --git a/archetype-validation-prober/README.md b/archetype-validation-prober/README.md new file mode 100644 index 00000000..cb1ed8f3 --- /dev/null +++ b/archetype-validation-prober/README.md @@ -0,0 +1,116 @@ +# Archetype Validation Prober + +A reference implementation of an **archetype validation gate** in front of Cloud Pub/Sub. + +It demonstrates a single principle: **only what minimally fits the process should ever be +enqueued.** Everything that is deterministically invalid is rejected *synchronously at the gate* and +never published, so it can never bounce in the queue, never inflate the dead-letter store, and never +burn engine resources being retried. + +The prober is provided as-is for demonstration purposes only, with no SLA. It is not meant to be run +as part of a production or critical workload. + +## Why + +A common failure mode of asynchronous messaging is treating a transport `200`/ack as if it meant the +message was *accepted*. It does not. A payload can be delivered and still be rejected on content, and +a message that never returns a positive functional ack **bounces forever** (or is silently dropped at +retention). At two messages that is noise; at two hundred thousand it is an outage. + +The fix is to separate the concerns and validate the deterministic ones **before** enqueuing: + +| Outcome | Nature | Where it is caught | Action | +| --- | --- | --- | --- | +| Conformant (functional ACK) | Success | — | Publish / `ack()` | +| Not JSON / wrong type / missing field / bad enum / bad pattern | **Deterministic** | **At the gate (this prober)** | **Reject synchronously, never enqueue** | +| Byte-different but semantically equal (charset / Unicode NFD) | False reject | At the gate (canonicalized) | Normalize to NFC, then accept | +| Downstream down (timeout, 5xx, connection) | Transient | Consumer | `nack()` → retry with backoff → dead-letter | +| State-dependent (unknown id, duplicate, business rule) | Unpredictable | Consumer | Route to quarantine / dead-letter | + +The gate handles the deterministic and false-reject rows. The consumer +([`ClassifyingReceiver`](src/main/java/com/google/cloud/pubsub/archetype/ClassifyingReceiver.java)) +handles the rest and, crucially, **only lets transient failures ride the redelivery machinery**; +functional rejects are `ack()`-ed and parked in quarantine so they never bounce. + +## Components + +- [`Archetype`](src/main/java/com/google/cloud/pubsub/archetype/Archetype.java) — the gate. Loads a + versioned JSON Schema (the *archetype*) and, cheapest check first, canonicalizes (charset + Unicode + NFC), then validates syntactically and structurally. Returns accept + canonical form, or reject + + machine-readable reason codes. Never throws on bad input. +- [`ValidationResult`](src/main/java/com/google/cloud/pubsub/archetype/ValidationResult.java) — the + terminal outcome (`ACCEPTED` with canonical payload, or `REJECTED` with reasons). +- [`ClassifyingReceiver`](src/main/java/com/google/cloud/pubsub/archetype/ClassifyingReceiver.java) — + a subscriber that classifies each processing failure (accepted / transient / functional reject) + and acts accordingly, keeping poison messages out of the retry loop. +- [`ArchetypeValidationGateway`](src/main/java/com/google/cloud/pubsub/archetype/ArchetypeValidationGateway.java) + — the runnable entry point. With no arguments it runs a self-contained, offline demo of the gate + over representative payloads. +- [`archetype.schema.json`](src/main/resources/archetype.schema.json) — the canonical contract used + by the demo. Replace it with your own to model your payload. + +## Build + +These instructions assume [Maven](https://maven.apache.org/) 3 and Java 8. + +``` +cd archetype-validation-prober +mvn package +``` + +The resulting jar is at `target/pubsub-archetype-validation-prober.jar`. + +## Run (offline demo) + +``` +java -jar target/pubsub-archetype-validation-prober.jar +``` + +Expected output (one line per sample): the conformant and the NFD-normalized payloads are +`ACCEPTED (enqueued)`; every deterministic failure is `REJECT` with its reason codes and is never +enqueued. + +### Options + +| Property | Type | Default | Description | +| --- | --- | --- | --- | +| `--archetype` | String | bundled `archetype.schema.json` | Path to the JSON Schema archetype to validate against. | +| `--charset` | String | `UTF-8` | Declared charset of incoming payloads, used for canonicalization. | +| `--help` | flag | — | Print usage and exit. | + +## Test + +``` +mvn test +``` + +## Smoke test (real Pub/Sub or emulator) + +The smoke tests are intentionally off by default. They validate the end-to-end +flow against either a local emulator or a real GCP project. + +### Emulator + +```bash +export PUBSUB_EMULATOR_HOST=localhost:8085 +gcloud beta emulators pubsub start +mvn verify -Psmoke +``` + +### Real GCP + +```bash +export GOOGLE_CLOUD_PROJECT=portfolioadvanced-llm +gcloud auth login +gcloud auth application-default login +mvn verify -Psmoke +``` + +The smoke suite covers: + +- valid payload round-trip +- invalid payload never reaches publish +- encoding rejection before JSON parsing + +For the higher-level GCP-native companion that demonstrates the same idea in +Python, see [PYTHON_COMPANION.md](PYTHON_COMPANION.md). diff --git a/archetype-validation-prober/pom.xml b/archetype-validation-prober/pom.xml new file mode 100644 index 00000000..8b8d58c5 --- /dev/null +++ b/archetype-validation-prober/pom.xml @@ -0,0 +1,137 @@ + + 4.0.0 + com.google.cloud.pubsub.archetype + archetype-validation-prober + jar + 1.0-SNAPSHOT + archetype-validation-prober + http://maven.apache.org + + + 1.8 + 1.8 + UTF-8 + + + + + junit + junit + 4.13.1 + test + + + com.beust + jcommander + 1.72 + + + com.google.cloud + google-cloud-pubsub + 1.116.3 + + + + com.networknt + json-schema-validator + 1.0.87 + + + com.fasterxml.jackson.core + jackson-databind + 2.13.5 + + + + + + + org.apache.maven.plugins + maven-shade-plugin + 2.3 + + true + + + *:* + + META-INF/*.SF + META-INF/*.DSA + META-INF/*.RSA + + + + + + + package + + shade + + + pubsub-archetype-validation-prober + + + com.google.cloud.pubsub.archetype.ArchetypeValidationGateway + + + + + + + + + org.apache.maven.plugins + maven-compiler-plugin + + 1.8 + 1.8 + + + + + + + + + smoke + + + + org.apache.maven.plugins + maven-failsafe-plugin + 3.2.5 + + + **/*SmokeIT.java + + + ${env.GOOGLE_CLOUD_PROJECT} + ${env.PUBSUB_EMULATOR_HOST} + + + + + + integration-test + verify + + + + + + + + + + diff --git a/archetype-validation-prober/src/main/java/com/google/cloud/pubsub/archetype/Archetype.java b/archetype-validation-prober/src/main/java/com/google/cloud/pubsub/archetype/Archetype.java new file mode 100644 index 00000000..9fd0462e --- /dev/null +++ b/archetype-validation-prober/src/main/java/com/google/cloud/pubsub/archetype/Archetype.java @@ -0,0 +1,132 @@ +// Copyright 2024 Google Inc. +// +// 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 +// +// http://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.google.cloud.pubsub.archetype; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.networknt.schema.JsonSchema; +import com.networknt.schema.JsonSchemaFactory; +import com.networknt.schema.SpecVersion; +import com.networknt.schema.ValidationMessage; +import java.io.IOException; +import java.io.InputStream; +import java.nio.ByteBuffer; +import java.nio.charset.CharacterCodingException; +import java.nio.charset.Charset; +import java.nio.charset.CharsetDecoder; +import java.nio.charset.CodingErrorAction; +import java.text.Normalizer; +import java.util.ArrayList; +import java.util.List; +import java.util.Set; + +/** + * The ingestion gate. + * + *

An {@code Archetype} is the single, canonical, versioned contract that every incoming payload + * must satisfy before it is ever published to a topic. It performs, cheapest check first: + * + *

    + *
  1. Canonicalization — decode with the declared charset and normalize to Unicode + * NFC, so byte-different-but-semantically-equal payloads (e.g. {@code ñ} as one code point vs + * {@code n} + combining tilde) do not produce false rejects. + *
  2. Syntactic — is it parseable JSON at all? + *
  3. Structural — does it match the archetype schema (required fields, types, + * enums, patterns, cardinality)? + *
+ * + *

Whatever passes is guaranteed to at minimum fit the process: the downstream engine + * will not blow up on a missing field or a type mismatch. Whatever fails is rejected + * deterministically with reason codes and is never enqueued. This deliberately keeps the error + * store (dead-letter / quarantine) small: it only ever holds the failures that genuinely cannot be + * predicted at the gate (downstream outages, state-dependent business rejections). + */ +public final class Archetype { + + private final JsonSchema schema; + private final ObjectMapper mapper = new ObjectMapper(); + + /** Loads the archetype from a JSON Schema (Draft 7) stream. */ + public Archetype(InputStream schemaStream) { + JsonSchemaFactory factory = JsonSchemaFactory.getInstance(SpecVersion.VersionFlag.V7); + this.schema = factory.getSchema(schemaStream); + } + + /** + * Loads the archetype from a classpath resource (e.g. {@code "/archetype.schema.json"}). + * Convenience factory for tests and the offline demo. + */ + public static Archetype fromResource(String resourcePath) { + InputStream stream = Archetype.class.getResourceAsStream(resourcePath); + if (stream == null) { + throw new IllegalArgumentException("Classpath resource not found: " + resourcePath); + } + return new Archetype(stream); + } + + /** + * Validates and canonicalizes a raw payload at the gate. + * + * @param rawBytes the payload exactly as received on the wire + * @param declaredCharset the charset the emitter claims to have used + * @return {@link ValidationResult#accepted} with the canonical form, or + * {@link ValidationResult#rejected} with reason codes; never throws for invalid input. + */ + public ValidationResult validate(byte[] rawBytes, Charset declaredCharset) { + // 1. Canonicalization: decode + normalize (Unicode NFC). Kills false rejects up front. + String canonical; + try { + CharsetDecoder decoder = declaredCharset.newDecoder(); + decoder.onMalformedInput(CodingErrorAction.REPORT); + decoder.onUnmappableCharacter(CodingErrorAction.REPORT); + String decoded = decoder.decode(ByteBuffer.wrap(rawBytes)).toString(); + canonical = Normalizer.normalize(decoded, Normalizer.Form.NFC); + } catch (CharacterCodingException | RuntimeException e) { + return ValidationResult.rejected( + single("ENCODING_UNDECODABLE: payload is not valid " + declaredCharset.name())); + } + + // 2. Syntactic: is it parseable JSON? + JsonNode node; + try { + node = mapper.readTree(canonical); + } catch (IOException e) { + return ValidationResult.rejected(single("SYNTAX_NOT_JSON: " + e.getMessage())); + } + if (node == null || node.isMissingNode()) { + return ValidationResult.rejected(single("SYNTAX_EMPTY: payload contained no JSON value")); + } + + // 3. Structural: does it satisfy the archetype schema? + Set violations = schema.validate(node); + if (!violations.isEmpty()) { + List reasons = new ArrayList<>(violations.size()); + for (ValidationMessage v : violations) { + // e.g. "CONTRACT_VIOLATION[$.policyNumber]: string found, integer expected" + reasons.add("CONTRACT_VIOLATION[" + v.getPath() + "]: " + v.getMessage()); + } + return ValidationResult.rejected(reasons); + } + + return ValidationResult.accepted(canonical); + } + + private static List single(String reason) { + List list = new ArrayList<>(1); + list.add(reason); + return list; + } +} diff --git a/archetype-validation-prober/src/main/java/com/google/cloud/pubsub/archetype/ArchetypeValidationGateway.java b/archetype-validation-prober/src/main/java/com/google/cloud/pubsub/archetype/ArchetypeValidationGateway.java new file mode 100644 index 00000000..97f6f58f --- /dev/null +++ b/archetype-validation-prober/src/main/java/com/google/cloud/pubsub/archetype/ArchetypeValidationGateway.java @@ -0,0 +1,151 @@ +// Copyright 2024 Google Inc. +// +// 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 +// +// http://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.google.cloud.pubsub.archetype; + +import com.beust.jcommander.JCommander; +import com.beust.jcommander.Parameter; +import java.io.FileInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.Charset; +import java.nio.charset.StandardCharsets; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.logging.Logger; + +/** + * Reference implementation of an archetype validation gate in front of Cloud Pub/Sub. + * + *

It demonstrates the principle that only what minimally fits the process should ever be + * enqueued. Every payload is validated and canonicalized against a single, versioned archetype + * (JSON Schema) at the gate: + * + *

+ * + *

Provided as-is for demonstration purposes only, with no SLA. Not meant to run as part of a + * production or critical workload. Run without arguments for a self-contained, offline demo of the + * gate over a set of representative payloads. + */ +public final class ArchetypeValidationGateway { + + private static final Logger logger = Logger.getLogger(ArchetypeValidationGateway.class.getName()); + + public static final class Args { + @Parameter( + names = "--archetype", + description = + "Path to the JSON Schema archetype. Defaults to the bundled archetype.schema.json.") + private String archetypePath = null; + + @Parameter( + names = "--charset", + description = "Declared charset of incoming payloads used for canonicalization.") + private String charset = "UTF-8"; + + @Parameter(names = "--help", help = true, description = "Print usage and exit.") + private boolean help = false; + } + + public static void main(String[] argv) throws IOException { + Args args = new Args(); + JCommander jc = JCommander.newBuilder().addObject(args).build(); + jc.parse(argv); + if (args.help) { + jc.usage(); + return; + } + + Charset charset = Charset.forName(args.charset); + Archetype archetype = new Archetype(openArchetype(args.archetypePath)); + + logger.info("Archetype validation gate ready. Running offline demo over sample payloads.\n"); + for (Map.Entry sample : samplePayloads().entrySet()) { + ValidationResult result = archetype.validate(sample.getValue(), charset); + if (result.isAccepted()) { + // In a live run this is where Publisher.publish(...) would be called. + logger.info(String.format("[PUBLISH ] %-24s -> ACCEPTED (enqueued)", sample.getKey())); + } else { + // Rejected at the gate: reported to the emitter, never enqueued. + logger.info( + String.format( + "[REJECT ] %-24s -> %s", sample.getKey(), result.reasons())); + } + } + } + + private static InputStream openArchetype(String path) throws IOException { + if (path != null) { + return new FileInputStream(path); + } + InputStream bundled = + ArchetypeValidationGateway.class.getResourceAsStream("/archetype.schema.json"); + if (bundled == null) { + throw new IOException("Bundled archetype.schema.json not found on classpath"); + } + return bundled; + } + + /** + * Representative payloads: a valid one, and the deterministic failure classes the gate is meant + * to stop before they ever reach the queue. + */ + private static Map samplePayloads() { + Map samples = new LinkedHashMap<>(); + + samples.put( + "valid", + utf8("{\"policyNumber\":\"POL-000123\",\"amount\":150.5,\"channel\":\"WEB\"}")); + + // Structural: wrong type (amount as string) and missing required field are contract violations. + samples.put( + "wrong-type", + utf8("{\"policyNumber\":\"POL-000123\",\"amount\":\"150.5\",\"channel\":\"WEB\"}")); + samples.put( + "missing-required", + utf8("{\"amount\":10.0,\"channel\":\"WEB\"}")); + + // Structural: value outside the declared enum. + samples.put( + "bad-enum", + utf8("{\"policyNumber\":\"POL-000999\",\"amount\":1.0,\"channel\":\"CARRIER_PIGEON\"}")); + + // Structural: pattern violation on the identifier. + samples.put( + "bad-pattern", + utf8("{\"policyNumber\":\"nope\",\"amount\":1.0,\"channel\":\"WEB\"}")); + + // Syntactic: not JSON at all. + samples.put("not-json", utf8("not a json payload")); + + // Canonicalization: 'ñ' expressed as NFD (n + U+0303). Semantically equal to NFC; the gate + // normalizes it so it does NOT become a false reject. + samples.put( + "nfd-normalization", + utf8("{\"policyNumber\":\"POL-000123\",\"amount\":1.0,\"channel\":\"WEB\",\"name\":\"Nun\u0303ez\"}")); + + return samples; + } + + private static byte[] utf8(String s) { + return s.getBytes(StandardCharsets.UTF_8); + } + + private ArchetypeValidationGateway() {} +} diff --git a/archetype-validation-prober/src/main/java/com/google/cloud/pubsub/archetype/ClassifyingReceiver.java b/archetype-validation-prober/src/main/java/com/google/cloud/pubsub/archetype/ClassifyingReceiver.java new file mode 100644 index 00000000..4857e48e --- /dev/null +++ b/archetype-validation-prober/src/main/java/com/google/cloud/pubsub/archetype/ClassifyingReceiver.java @@ -0,0 +1,121 @@ +// Copyright 2024 Google Inc. +// +// 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 +// +// http://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.google.cloud.pubsub.archetype; + +import com.google.cloud.pubsub.v1.AckReplyConsumer; +import com.google.cloud.pubsub.v1.MessageReceiver; +import com.google.cloud.pubsub.v1.Publisher; +import com.google.protobuf.ByteString; +import com.google.pubsub.v1.PubsubMessage; +import java.util.logging.Level; +import java.util.logging.Logger; + +/** + * A subscriber that classifies every processing failure before deciding what to do with the + * message. This is the piece that stops poison messages from bouncing forever. + * + *

The redelivery machinery of Pub/Sub (nack → retry → dead-letter) is only for + * transient failures. Deterministic failures must never ride it: retrying them is pointless and, at + * scale, self-inflicts a redelivery storm. Hence the three-way decision: + * + * + * + * + * + * + * + * + * + *
Delivery decision table
Downstream outcomeNatureAction
Accepted (functional ACK)Success{@code ack()} — done
Transient (timeout, 5xx, connection)Retryable{@code nack()} — redeliver with backoff, eventually dead-letter
Functional reject (contract / data)Deterministic{@code ack()} + republish to quarantine — never retried
+ */ +public final class ClassifyingReceiver implements MessageReceiver { + + private static final Logger logger = Logger.getLogger(ClassifyingReceiver.class.getName()); + + /** How a (simulated or real) downstream delivery ended. */ + public enum DeliveryOutcome { + ACCEPTED, + TRANSIENT_FAILURE, + FUNCTIONAL_REJECT + } + + /** Pluggable downstream. Real deployments wire this to the actual receiving system. */ + public interface Downstream { + DeliveryOutcome deliver(PubsubMessage message); + } + + private final Downstream downstream; + private final Publisher quarantinePublisher; + + public ClassifyingReceiver(Downstream downstream, Publisher quarantinePublisher) { + this.downstream = downstream; + this.quarantinePublisher = quarantinePublisher; + } + + @Override + public void receiveMessage(PubsubMessage message, AckReplyConsumer consumer) { + DeliveryOutcome outcome; + try { + outcome = downstream.deliver(message); + } catch (RuntimeException unexpected) { + // Unknown exceptions are treated as transient: better to retry than to silently drop. + logger.log(Level.WARNING, "Downstream threw; treating as transient", unexpected); + consumer.nack(); + return; + } + + switch (outcome) { + case ACCEPTED: + consumer.ack(); + return; + + case TRANSIENT_FAILURE: + // Retryable: let Pub/Sub redeliver with backoff (and eventually dead-letter it). + consumer.nack(); + return; + + case FUNCTIONAL_REJECT: + default: + // Deterministic: it will fail identically forever. Remove it from the retry loop + // (ack) and park it in quarantine with its reason for a human/spec decision. + quarantine(message); + consumer.ack(); + return; + } + } + + private void quarantine(PubsubMessage original) { + PubsubMessage tagged = + PubsubMessage.newBuilder() + .setData(original.getData()) + .putAllAttributes(original.getAttributesMap()) + .putAttributes("quarantine-reason", "FUNCTIONAL_REJECT") + .putAttributes("original-message-id", original.getMessageId()) + .build(); + try { + quarantinePublisher.publish(tagged).get(); + } catch (Exception e) { + // If we cannot even quarantine, fall back to nack so the message is not lost. + logger.log(Level.SEVERE, "Failed to publish to quarantine topic", e); + throw new RuntimeException(e); + } + } + + /** Convenience for callers that hold raw bytes rather than a PubsubMessage. */ + static PubsubMessage message(String data) { + return PubsubMessage.newBuilder().setData(ByteString.copyFromUtf8(data)).build(); + } +} diff --git a/archetype-validation-prober/src/main/java/com/google/cloud/pubsub/archetype/ValidationResult.java b/archetype-validation-prober/src/main/java/com/google/cloud/pubsub/archetype/ValidationResult.java new file mode 100644 index 00000000..0d97cf2e --- /dev/null +++ b/archetype-validation-prober/src/main/java/com/google/cloud/pubsub/archetype/ValidationResult.java @@ -0,0 +1,81 @@ +// Copyright 2024 Google Inc. +// +// 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 +// +// http://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.google.cloud.pubsub.archetype; + +import java.util.Collections; +import java.util.List; + +/** + * Outcome of validating an incoming payload against the archetype at the ingestion gate. + * + *

A payload is either {@link Status#ACCEPTED} (structurally conformant and therefore safe to + * publish, carrying its canonicalized form) or {@link Status#REJECTED} (deterministically invalid, + * carrying machine-readable reason codes). A rejected payload is never enqueued: retrying it would + * fail identically forever, so it is reported synchronously to the emitter instead. + */ +public final class ValidationResult { + + /** Terminal classification of a payload at the gate. */ + public enum Status { + ACCEPTED, + REJECTED + } + + private final Status status; + private final String canonicalPayload; + private final List reasons; + + private ValidationResult(Status status, String canonicalPayload, List reasons) { + this.status = status; + this.canonicalPayload = canonicalPayload; + this.reasons = reasons; + } + + /** Builds an accepted result carrying the canonicalized (normalized) payload. */ + public static ValidationResult accepted(String canonicalPayload) { + return new ValidationResult(Status.ACCEPTED, canonicalPayload, Collections.emptyList()); + } + + /** Builds a rejected result carrying one or more reason codes describing why it was refused. */ + public static ValidationResult rejected(List reasons) { + return new ValidationResult(Status.REJECTED, null, Collections.unmodifiableList(reasons)); + } + + public Status status() { + return status; + } + + public boolean isAccepted() { + return status == Status.ACCEPTED; + } + + /** The canonicalized payload, only present when {@link #isAccepted()} is true. */ + public String canonicalPayload() { + return canonicalPayload; + } + + /** Machine-readable reason codes, only populated when the payload was rejected. */ + public List reasons() { + return reasons; + } + + @Override + public String toString() { + return status == Status.ACCEPTED + ? "ACCEPTED" + : "REJECTED" + reasons; + } +} diff --git a/archetype-validation-prober/src/main/resources/archetype.schema.json b/archetype-validation-prober/src/main/resources/archetype.schema.json new file mode 100644 index 00000000..60b1ab57 --- /dev/null +++ b/archetype-validation-prober/src/main/resources/archetype.schema.json @@ -0,0 +1,31 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "$id": "https://example.com/archetypes/ingest-message/v1.json", + "title": "IngestMessage", + "description": "Canonical, versioned archetype every incoming payload must satisfy at the gate before it is published. Whatever passes this schema is guaranteed to at minimum fit the downstream process.", + "type": "object", + "additionalProperties": false, + "required": ["policyNumber", "amount", "channel"], + "properties": { + "policyNumber": { + "type": "string", + "description": "Business identifier. Fixed format so both emitter and receiver agree byte-for-byte.", + "pattern": "^POL-\\d{6}$" + }, + "amount": { + "type": "number", + "description": "Monetary amount. Must be a JSON number, never a string.", + "minimum": 0 + }, + "channel": { + "type": "string", + "description": "Origin channel. Closed enum: anything else is a deterministic reject.", + "enum": ["WEB", "APP", "IVR", "BRANCH"] + }, + "name": { + "type": "string", + "description": "Optional free-text name. Canonicalized to Unicode NFC at the gate.", + "maxLength": 140 + } + } +} diff --git a/archetype-validation-prober/src/test/java/com/google/cloud/pubsub/archetype/ArchetypeTest.java b/archetype-validation-prober/src/test/java/com/google/cloud/pubsub/archetype/ArchetypeTest.java new file mode 100644 index 00000000..5ecc24a7 --- /dev/null +++ b/archetype-validation-prober/src/test/java/com/google/cloud/pubsub/archetype/ArchetypeTest.java @@ -0,0 +1,107 @@ +// Copyright 2024 Google Inc. +// +// 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 +// +// http://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.google.cloud.pubsub.archetype; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import org.junit.Before; +import org.junit.Test; + +/** Unit tests for the {@link Archetype} ingestion gate. */ +public class ArchetypeTest { + + private Archetype archetype; + + @Before + public void setUp() { + InputStream schema = getClass().getResourceAsStream("/archetype.schema.json"); + archetype = new Archetype(schema); + } + + private ValidationResult validate(String json) { + return archetype.validate(json.getBytes(StandardCharsets.UTF_8), StandardCharsets.UTF_8); + } + + @Test + public void acceptsConformantPayload() { + ValidationResult r = validate("{\"policyNumber\":\"POL-000123\",\"amount\":150.5,\"channel\":\"WEB\"}"); + assertTrue(r.reasons().toString(), r.isAccepted()); + } + + @Test + public void rejectsWrongType() { + ValidationResult r = validate("{\"policyNumber\":\"POL-000123\",\"amount\":\"150.5\",\"channel\":\"WEB\"}"); + assertFalse(r.isAccepted()); + assertTrue(r.reasons().toString().contains("amount")); + } + + @Test + public void rejectsMissingRequired() { + ValidationResult r = validate("{\"amount\":10.0,\"channel\":\"WEB\"}"); + assertFalse(r.isAccepted()); + assertTrue(r.reasons().toString().contains("policyNumber")); + } + + @Test + public void rejectsValueOutsideEnum() { + ValidationResult r = validate("{\"policyNumber\":\"POL-000999\",\"amount\":1.0,\"channel\":\"CARRIER_PIGEON\"}"); + assertFalse(r.isAccepted()); + } + + @Test + public void rejectsPatternViolation() { + ValidationResult r = validate("{\"policyNumber\":\"nope\",\"amount\":1.0,\"channel\":\"WEB\"}"); + assertFalse(r.isAccepted()); + } + + @Test + public void rejectsNonJson() { + ValidationResult r = validate("not json"); + assertFalse(r.isAccepted()); + assertTrue(r.reasons().toString().contains("SYNTAX")); + } + + @Test + public void rejectsMalformedUtf8Bytes() { + byte[] malformed = new byte[] {(byte) 0xC3, (byte) 0x28}; + ValidationResult r = archetype.validate(malformed, StandardCharsets.UTF_8); + assertFalse(r.isAccepted()); + assertTrue(r.reasons().toString(), r.reasons().get(0).startsWith("ENCODING_UNDECODABLE")); + } + + @Test + public void canonicalizesNfdToNfcSoItIsNotAFalseReject() { + // "Nuñez" with 'ñ' as NFD (n + U+0303). Must be accepted and stored as NFC. + String nfd = "{\"policyNumber\":\"POL-000123\",\"amount\":1.0,\"channel\":\"WEB\",\"name\":\"Nun\u0303ez\"}"; + ValidationResult r = validate(nfd); + assertTrue(r.reasons().toString(), r.isAccepted()); + // NFC form of the name ("Nuñez") must be present; the NFD combining sequence must be gone. + assertTrue(r.canonicalPayload().contains("Nu\u00F1ez")); + assertFalse(r.canonicalPayload().contains("n\u0303")); + } + + @Test + public void reportsMultipleReasonsAtOnce() { + ValidationResult r = validate("{\"policyNumber\":\"nope\",\"channel\":\"CARRIER_PIGEON\"}"); + assertFalse(r.isAccepted()); + // At least the missing 'amount' and the bad enum/pattern should be reported. + assertTrue(r.reasons().size() >= 2); + } +} diff --git a/archetype-validation-prober/src/test/java/com/google/cloud/pubsub/archetype/smoke/ArchetypeGateSmokeIT.java b/archetype-validation-prober/src/test/java/com/google/cloud/pubsub/archetype/smoke/ArchetypeGateSmokeIT.java new file mode 100644 index 00000000..37b9f119 --- /dev/null +++ b/archetype-validation-prober/src/test/java/com/google/cloud/pubsub/archetype/smoke/ArchetypeGateSmokeIT.java @@ -0,0 +1,245 @@ +// Copyright 2024 Google Inc. +// +// 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 +// +// http://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.google.cloud.pubsub.archetype.smoke; + +import com.google.api.gax.core.NoCredentialsProvider; +import com.google.api.gax.grpc.GrpcTransportChannel; +import com.google.api.gax.rpc.FixedTransportChannelProvider; +import com.google.api.gax.rpc.TransportChannelProvider; +import com.google.cloud.pubsub.v1.Publisher; +import com.google.cloud.pubsub.v1.SubscriptionAdminClient; +import com.google.cloud.pubsub.v1.TopicAdminClient; +import com.google.cloud.pubsub.v1.stub.GrpcSubscriberStub; +import com.google.cloud.pubsub.v1.stub.SubscriberStubSettings; +import com.google.cloud.pubsub.archetype.Archetype; +import com.google.cloud.pubsub.archetype.ValidationResult; +import com.google.protobuf.ByteString; +import com.google.pubsub.v1.ProjectSubscriptionName; +import com.google.pubsub.v1.ProjectTopicName; +import com.google.pubsub.v1.PubsubMessage; +import com.google.pubsub.v1.PullRequest; +import com.google.pubsub.v1.PullResponse; +import com.google.pubsub.v1.PushConfig; +import io.grpc.ManagedChannel; +import io.grpc.ManagedChannelBuilder; +import org.junit.After; +import org.junit.Assume; +import org.junit.Before; +import org.junit.Test; + +import java.nio.charset.StandardCharsets; +import java.util.UUID; +import java.util.concurrent.TimeUnit; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +/** + * Smoke tests for {@link Archetype} against a real or emulated Pub/Sub endpoint. + * + *

These tests are intentionally off by default. They run only under the {@code smoke} + * Maven profile ({@code mvn verify -Psmoke}) so they never execute in a standard unit build. + * + *

Two modes are supported: + *

    + *
  1. Emulator (default, zero cost): set {@code PUBSUB_EMULATOR_HOST=localhost:8085} + * and start the emulator with {@code gcloud beta emulators pubsub start}. + *
  2. Real GCP topic: set {@code GOOGLE_CLOUD_PROJECT}, {@code PUBSUB_SMOKE_TOPIC}, + * and {@code PUBSUB_SMOKE_SUBSCRIPTION}. Application Default Credentials must be valid. + *
+ * + *

Topic and subscription are created and deleted per-run; the test is hermetic and leaves no + * residue in the project. + * + *

What these tests prove

+ * + */ +public class ArchetypeGateSmokeIT { + + // ── Environment resolution ────────────────────────────────────────────── + + private static final String ENV_PROJECT = System.getenv("GOOGLE_CLOUD_PROJECT"); + private static final String ENV_EMULATOR = System.getenv("PUBSUB_EMULATOR_HOST"); + private static final String TOPIC_ID = "archetype-smoke-" + UUID.randomUUID(); + private static final String SUBSCRIPTION_ID = TOPIC_ID + "-sub"; + + private static final String VALID_PAYLOAD = + "{\"event_type\":\"llm_request\",\"payload\":{\"prompt\":\"smoke\"},\"version\":\"1.0\"}"; + + private TopicAdminClient topicAdmin; + private SubscriptionAdminClient subscriptionAdmin; + private Publisher publisher; + private GrpcSubscriberStub subscriberStub; + private ManagedChannel channel; + private Archetype gate; + + // ── Setup / teardown ──────────────────────────────────────────────────── + + @Before + public void setUp() throws Exception { + // Skip if neither emulator nor real project is configured. + Assume.assumeTrue( + "Set PUBSUB_EMULATOR_HOST or GOOGLE_CLOUD_PROJECT to run smoke tests", + ENV_EMULATOR != null || ENV_PROJECT != null); + + gate = Archetype.fromResource("/archetype.schema.json"); + + if (ENV_EMULATOR != null) { + channel = ManagedChannelBuilder.forTarget(ENV_EMULATOR).usePlaintext().build(); + TransportChannelProvider channelProvider = + FixedTransportChannelProvider.create(GrpcTransportChannel.create(channel)); + NoCredentialsProvider credentialsProvider = NoCredentialsProvider.create(); + + topicAdmin = TopicAdminClient.create( + com.google.cloud.pubsub.v1.TopicAdminSettings.newBuilder() + .setTransportChannelProvider(channelProvider) + .setCredentialsProvider(credentialsProvider) + .build()); + subscriptionAdmin = SubscriptionAdminClient.create( + com.google.cloud.pubsub.v1.SubscriptionAdminSettings.newBuilder() + .setTransportChannelProvider(channelProvider) + .setCredentialsProvider(credentialsProvider) + .build()); + + publisher = Publisher.newBuilder(ProjectTopicName.of("smoke-project", TOPIC_ID)) + .setChannelProvider(channelProvider) + .setCredentialsProvider(credentialsProvider) + .build(); + + SubscriberStubSettings stubSettings = SubscriberStubSettings.newBuilder() + .setTransportChannelProvider(channelProvider) + .setCredentialsProvider(credentialsProvider) + .build(); + subscriberStub = GrpcSubscriberStub.create(stubSettings); + + topicAdmin.createTopic(ProjectTopicName.of("smoke-project", TOPIC_ID)); + subscriptionAdmin.createSubscription( + ProjectSubscriptionName.of("smoke-project", SUBSCRIPTION_ID), + ProjectTopicName.of("smoke-project", TOPIC_ID), + PushConfig.getDefaultInstance(), 10); + } else { + // Real GCP — use ADC + topicAdmin = TopicAdminClient.create(); + subscriptionAdmin = SubscriptionAdminClient.create(); + publisher = Publisher.newBuilder(ProjectTopicName.of(ENV_PROJECT, TOPIC_ID)).build(); + subscriberStub = GrpcSubscriberStub.create(SubscriberStubSettings.newBuilder().build()); + + topicAdmin.createTopic(ProjectTopicName.of(ENV_PROJECT, TOPIC_ID)); + subscriptionAdmin.createSubscription( + ProjectSubscriptionName.of(ENV_PROJECT, SUBSCRIPTION_ID), + ProjectTopicName.of(ENV_PROJECT, TOPIC_ID), + PushConfig.getDefaultInstance(), 10); + } + } + + @After + public void tearDown() throws Exception { + if (publisher != null) publisher.shutdown(); + String project = ENV_EMULATOR != null ? "smoke-project" : ENV_PROJECT; + if (subscriptionAdmin != null) { + try { subscriptionAdmin.deleteSubscription( + ProjectSubscriptionName.of(project, SUBSCRIPTION_ID)); } catch (Exception ignored) {} + subscriptionAdmin.close(); + } + if (topicAdmin != null) { + try { topicAdmin.deleteTopic(ProjectTopicName.of(project, TOPIC_ID)); } catch (Exception ignored) {} + topicAdmin.close(); + } + if (subscriberStub != null) subscriberStub.close(); + if (channel != null) channel.shutdownNow(); + } + + // ── Tests ─────────────────────────────────────────────────────────────── + + /** + * A conformant payload passes the gate, is published, and is pulled back byte-identical. + * Proves round-trip integrity through the real (or emulated) Pub/Sub transport. + */ + @Test + public void validPayload_passesGateAndRoundTrips() throws Exception { + byte[] raw = VALID_PAYLOAD.getBytes(StandardCharsets.UTF_8); + ValidationResult result = gate.validate(raw, StandardCharsets.UTF_8); + assertTrue("Gate should accept a conformant payload", result.isAccepted()); + + // Publish the canonical form + String canonical = result.getCanonical(); + publisher.publish(PubsubMessage.newBuilder() + .setData(ByteString.copyFromUtf8(canonical)) + .build()).get(10, TimeUnit.SECONDS); + + // Pull back and verify byte equality + String project = ENV_EMULATOR != null ? "smoke-project" : ENV_PROJECT; + PullResponse pullResponse = subscriberStub.pullCallable().call( + PullRequest.newBuilder() + .setSubscription(ProjectSubscriptionName.of(project, SUBSCRIPTION_ID).toString()) + .setMaxMessages(1) + .build()); + + assertFalse("Expected at least one message", pullResponse.getReceivedMessagesList().isEmpty()); + String pulled = pullResponse.getReceivedMessages(0).getMessage().getData().toStringUtf8(); + assertEquals("Round-trip payload must be byte-identical to canonical form", canonical, pulled); + } + + /** + * An invalid payload (schema violation) is rejected by the gate and never reaches publish(). + * Verifies the gate holds the line against a real topic — the topic should remain empty. + */ + @Test + public void invalidPayload_rejectedAtGate_neverPublished() throws Exception { + byte[] badPayload = "{\"event_type\":\"unknown\",\"payload\":{}}".getBytes(StandardCharsets.UTF_8); + ValidationResult result = gate.validate(badPayload, StandardCharsets.UTF_8); + + assertFalse("Gate should reject a payload with invalid enum and missing required field", + result.isAccepted()); + assertFalse("Rejection reasons must be non-empty", result.getRejectionReasons().isEmpty()); + + // Do NOT call publisher.publish() — verifying the caller respects the rejection. + // Pull from the subscription: it must be empty. + String project = ENV_EMULATOR != null ? "smoke-project" : ENV_PROJECT; + PullResponse pullResponse = subscriberStub.pullCallable().call( + PullRequest.newBuilder() + .setSubscription(ProjectSubscriptionName.of(project, SUBSCRIPTION_ID).toString()) + .setMaxMessages(1) + .build()); + + assertTrue("Topic must be empty — invalid payload must never reach publish()", + pullResponse.getReceivedMessagesList().isEmpty()); + } + + /** + * A payload with invalid UTF-8 bytes is rejected at the encoding stage, + * before any JSON parsing or schema validation is attempted. + */ + @Test + public void encodingError_rejectedBeforeJsonParsing() { + byte[] badBytes = new byte[]{(byte) 0xFF, (byte) 0xFE, 0x7B, 0x7D}; // invalid UTF-8 + {} + ValidationResult result = gate.validate(badBytes, StandardCharsets.UTF_8); + + assertFalse("Encoding error must be rejected", result.isAccepted()); + assertTrue("Rejection reason must mention encoding", + result.getRejectionReasons().stream() + .anyMatch(r -> r.contains("ENCODING_UNDECODABLE"))); + } +}