Review only: P1 committed mutation integrated candidate

README.md

@@ -322,6 +322,7 @@ Request JSON is not allowed to replace those trusted capabilities.
 - [Shared load state](examples/shared-load-state/README.md)
 - [Binary object storage and temporary S3 URLs](examples/object-storage/README.md)
 - [Context-owned HTTP with bounded retries and safe diagnostics](examples/http-resilience/README.md)
+- [Transactional committed-event Outbox from a real SQLite graph save](examples/committed-events/README.md)
 - [Extension points](https://github.com/teaql/teaql-conformance/blob/main/design/runtime-customization-p0-p1-matrix.md)
 
 ## Safe troubleshooting

examples/committed-events/README.md

@@ -0,0 +1,31 @@
+# SQLite committed-event source example
+
+Run `bash examples/committed-events/verify.sh`. It executes a small, real
+`Order.save(context)` flow twice against a local Java source. Each round uses
+fresh source/target SQLite files and an independent DuckDB target. The test
+reopens the Outbox, delivers the two events through bounded publishers into
+both model-checked targets, verifies their rows and the SQLite target duplicate
+acknowledgement, then injects a later source write failure. The first SQLite
+delivery deliberately loses its acknowledgement after target commit; a new
+source executor replays sequence 1 as a duplicate and then applies sequence 2
+in order. This small fixture
+itself requires no generated ERP model or remote service.
+
+The example configures `SqliteCommittedMutationRecorder` at trusted runtime
+startup; `context.ensureSchema()` installs its infrastructure tables. A database trigger
+sets `memo` after insert; the Outbox contains that authoritative readback, not
+the pre-save entity value. Subsequent update produces sequence 2. When the
+Outbox table is made unavailable, the attempted third save rolls back its
+business update and emits no committed event.
+
+This is **not** authenticated network replication. The in-process sinks prove
+local cursor/target behavior; transport identity, retry budgets, snapshot-plus-
+tail and dynamic fields remain separate work. The included typed wire tests
+cover selected scalar, null, date, instant and Decimal boundaries, not every
+possible model type.
+
+The verifier also runs an application-owned generated Standard Order consumer
+twice. It creates a Customer through generated audited Save on SQLite, delivers
+one committed set to a model-checked DuckDB target, and reads the target through
+generated Q. The checked-in generated library remains byte-for-byte unchanged;
+public-package and production-network qualification are separate.

examples/committed-events/pom.xml

@@ -0,0 +1,20 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <groupId>io.teaql</groupId><artifactId>teaql-java-parent</artifactId>
+    <version>1.555-RELEASE</version><relativePath>../../pom.xml</relativePath>
+  </parent>
+  <artifactId>teaql-committed-events-example</artifactId>
+  <dependencies>
+    <dependency><groupId>io.teaql</groupId><artifactId>teaql-runtime</artifactId></dependency>
+    <dependency><groupId>io.teaql</groupId><artifactId>teaql-sqlite</artifactId></dependency>
+    <dependency><groupId>io.teaql</groupId><artifactId>teaql-duckdb</artifactId><version>${project.version}</version></dependency>
+    <dependency><groupId>io.teaql</groupId><artifactId>teaql-provider-jdbc</artifactId></dependency>
+    <dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-databind</artifactId></dependency>
+    <dependency><groupId>org.xerial</groupId><artifactId>sqlite-jdbc</artifactId><version>3.53.2.1</version></dependency>
+    <dependency><groupId>org.duckdb</groupId><artifactId>duckdb_jdbc</artifactId><version>1.5.5.1</version></dependency>
+    <dependency><groupId>junit</groupId><artifactId>junit</artifactId><scope>test</scope></dependency>
+  </dependencies>
+</project>

examples/committed-events/src/test/java/io/teaql/examples/committed/CommittedEventsExampleTest.java

@@ -0,0 +1,223 @@
+package io.teaql.examples.committed;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.teaql.core.BaseEntity;
+import io.teaql.core.duck.DuckDbCommittedInbox;
+import io.teaql.core.duck.DuckDbModelCheckedCommittedTarget;
+import io.teaql.core.meta.SimpleEntityMetaFactory;
+import io.teaql.core.sql.GenericSQLProperty;
+import io.teaql.core.sql.SQLEntityDescriptor;
+import io.teaql.core.sqlite.SqliteCommittedEventStore;
+import io.teaql.core.sqlite.SqliteCommittedEventStore.InboxOutcome;
+import io.teaql.core.sqlite.SqliteCommittedMutationRecorder;
+import io.teaql.core.sqlite.SqliteModelCheckedCommittedTarget;
+import io.teaql.core.sqlite.SqliteCommittedOutboxPublisher;
+import io.teaql.core.sqlite.SqliteDataServiceExecutor;
+import io.teaql.provider.jdbc.JdbcSqlExecutor;
+import io.teaql.runtime.DefaultUserContext;
+import io.teaql.runtime.TeaQLRuntime;
+import org.junit.Test;
+import org.sqlite.SQLiteDataSource;
+
+import javax.sql.DataSource;
+import java.io.PrintWriter;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.SQLFeatureNotSupportedException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.logging.Logger;
+
+import static org.junit.Assert.*;
+
+public class CommittedEventsExampleTest {
+    private record DriverManagerDataSource(String url) implements DataSource {
+        @Override public Connection getConnection() throws SQLException { return DriverManager.getConnection(url); }
+        @Override public Connection getConnection(String user, String password) throws SQLException {
+            return getConnection();
+        }
+        @Override public PrintWriter getLogWriter() { return null; }
+        @Override public void setLogWriter(PrintWriter writer) { }
+        @Override public void setLoginTimeout(int seconds) { }
+        @Override public int getLoginTimeout() { return 0; }
+        @Override public Logger getParentLogger() throws SQLFeatureNotSupportedException {
+            throw new SQLFeatureNotSupportedException();
+        }
+        @Override public <T> T unwrap(Class<T> type) throws SQLException { throw new SQLException("not a wrapper"); }
+        @Override public boolean isWrapperFor(Class<?> type) { return false; }
+    }
+
+    public static final class Order extends BaseEntity {
+        private final Map<String, Object> fields = new HashMap<>();
+
+        @Override public String typeName() { return "Order"; }
+        @Override public Object __internalGet(String field) {
+            return field.equals("name") || field.equals("memo") ? fields.get(field) : super.__internalGet(field);
+        }
+        @Override public void __internalSet(String field, Object value) {
+            if (field.equals("name") || field.equals("memo")) fields.put(field, value);
+            else super.__internalSet(field, value);
+        }
+    }
+
+    @Test public void realSaveFreezesDatabaseResultsBeforeCommit() throws Exception {
+        Path path = Files.createTempFile("teaql-committed-example-", ".sqlite");
+        path.toFile().deleteOnExit();
+        SQLiteDataSource source = new SQLiteDataSource();
+        source.setUrl("jdbc:sqlite:" + path);
+        JdbcSqlExecutor sql = new JdbcSqlExecutor(source);
+        SqliteCommittedMutationRecorder recorder = new SqliteCommittedMutationRecorder(
+                sql, "example-java-source", "sha256:" + "0".repeat(64));
+
+        SQLEntityDescriptor descriptor = new SQLEntityDescriptor();
+        descriptor.setType("Order");
+        descriptor.setTargetType(Order.class);
+        descriptor.setEntitySupplier(Order::new);
+        descriptor.setDataService("sqlite");
+        for (String field : new String[] {"id", "version", "name", "memo"}) {
+            boolean text = field.equals("name") || field.equals("memo");
+            GenericSQLProperty property = (GenericSQLProperty) descriptor.addSimpleProperty(
+                    field, text ? String.class : Long.class);
+            property.setColumnType(text ? "VARCHAR(255)" : "BIGINT");
+        }
+        SimpleEntityMetaFactory metadata = new SimpleEntityMetaFactory();
+        metadata.register(descriptor);
+        TeaQLRuntime runtime = TeaQLRuntime.builder().metadata(metadata)
+                .dataService("sqlite", new SqliteDataServiceExecutor("sqlite", sql, source))
+                .idGenerationService((caller, entity) -> 42L)
+                .committedMutationRecorder(recorder)
+                .executionLogging(false).build();
+        DefaultUserContext context = new DefaultUserContext(runtime);
+        context.ensureSchema();
+        sql.execute("CREATE TRIGGER order_memo AFTER INSERT ON order_data "
+                + "BEGIN UPDATE order_data SET memo = 'set by database' WHERE id = NEW.id; END");
+
+        Order order = new Order();
+        order.__internalInitializeNewEntityId(42L);
+        order.updateProperty("name", "draft");
+        order.auditAs("create example order").save(context);
+        var first = recorder.eventStore().outboxAfter("example-java-source", 0, 10);
+        assertEquals(1, first.size());
+        assertEquals(1, first.get(0).sourceSequence());
+        JsonNode created = new ObjectMapper().readTree(first.get(0).payloadJson());
+        assertEquals("set by database", created.path("mutations").get(0).path("fields").path("memo").asText());
+        assertEquals("create example order", created.path("audit").path("comment").asText());
+
+        order.updateProperty("name", "submitted");
+        order.auditAs("submit example order").save(context);
+        JdbcSqlExecutor reopened = new JdbcSqlExecutor(source);
+        var durable = new SqliteCommittedEventStore(reopened)
+                .outboxAfter("example-java-source", 0, 10);
+        assertEquals(2, durable.size());
+        JsonNode submitted = new ObjectMapper().readTree(durable.get(1).payloadJson());
+        assertEquals("1", submitted.path("mutations").get(0).path("base_version").asText());
+        assertEquals("2", submitted.path("mutations").get(0).path("committed_version").asText());
+
+        Path replicaPath = Files.createTempFile("teaql-committed-target-example-", ".sqlite");
+        replicaPath.toFile().deleteOnExit();
+        SQLiteDataSource replicaSource = new SQLiteDataSource();
+        replicaSource.setUrl("jdbc:sqlite:" + replicaPath);
+        JdbcSqlExecutor replicaSql = new JdbcSqlExecutor(replicaSource);
+        TeaQLRuntime replicaRuntime = TeaQLRuntime.builder().metadata(metadata)
+                .dataService("sqlite", new SqliteDataServiceExecutor("sqlite", replicaSql, replicaSource))
+                .executionLogging(false).build();
+        DefaultUserContext replicaContext = new DefaultUserContext(replicaRuntime);
+        replicaContext.ensureSchema();
+        SqliteModelCheckedCommittedTarget target = new SqliteModelCheckedCommittedTarget(
+                replicaSql, "example-java-source", "sha256:" + "0".repeat(64));
+        target.ensureSchema();
+        var interruptedPublisher = new SqliteCommittedOutboxPublisher(reopened,
+                "example-java-source", "example-analytics", "sha256:" + "0".repeat(64));
+        assertThrows(IllegalStateException.class, () -> interruptedPublisher.publishBatch(10,
+                record -> {
+                    target.apply(replicaContext, record);
+                    throw new IllegalStateException("simulate lost target acknowledgement");
+                }));
+        assertEquals(0L, interruptedPublisher.status().ackedSequence());
+        assertEquals(2L, interruptedPublisher.status().backlog());
+        assertEquals(1, replicaSql.queryForList("SELECT * FROM teaql_event_inbox",
+                new Object[0]).size());
+
+        SQLiteDataSource restartedSource = new SQLiteDataSource();
+        restartedSource.setUrl("jdbc:sqlite:" + path);
+        JdbcSqlExecutor restarted = new JdbcSqlExecutor(restartedSource);
+        var published = new SqliteCommittedOutboxPublisher(restarted, "example-java-source",
+                "example-analytics", "sha256:" + "0".repeat(64))
+                .publishBatch(10, record -> target.apply(replicaContext, record));
+        assertEquals(2, published.acknowledged());
+        assertEquals(0, published.status().backlog());
+        assertEquals(2L, published.status().ackedSequence());
+        var duplicate = target.apply(replicaContext, durable.get(0));
+        assertTrue(duplicate.duplicate());
+        assertEquals(1, duplicate.checkpoint());
+        var replicated = replicaSql.queryForList("SELECT version, name, memo FROM order_data WHERE id = ?",
+                new Object[] {42L}).get(0);
+        assertEquals(2L, ((Number) replicated.get("version")).longValue());
+        assertEquals("submitted", replicated.get("name"));
+        assertEquals("set by database", replicated.get("memo"));
+        assertEquals(2, replicaSql.queryForList("SELECT * FROM teaql_event_applied_payload",
+                new Object[0]).size());
+        System.out.println("COMMITTED_DELIVERY_RECOVERY_PASS language=java ordered=2 "
+                + "lost_ack=1 restart=1 duplicate=1");
+        replicaRuntime.close();
+
+        Path duckDirectory = Files.createTempDirectory("teaql-committed-duckdb-example-");
+        duckDirectory.toFile().deleteOnExit();
+        Path duckPath = duckDirectory.resolve("target.db");
+        duckPath.toFile().deleteOnExit();
+        JdbcSqlExecutor duckSql = new JdbcSqlExecutor(
+                new DriverManagerDataSource("jdbc:duckdb:" + duckPath));
+        duckSql.execute("CREATE TABLE order_data (id BIGINT PRIMARY KEY, version BIGINT NOT NULL, "
+                + "name VARCHAR, memo VARCHAR)");
+        DuckDbModelCheckedCommittedTarget duckTarget =
+                new DuckDbModelCheckedCommittedTarget(duckSql, "example-java-source",
+                        "sha256:" + "0".repeat(64));
+        duckTarget.ensureInboxSchema();
+        var duckPublished = new SqliteCommittedOutboxPublisher(reopened, "example-java-source",
+                "example-duckdb-analytics", "sha256:" + "0".repeat(64))
+                .publishBatch(10, record -> {
+                    var outcome = duckTarget.apply(context, new DuckDbCommittedInbox.Record(
+                            record.sourceRuntimeId(), record.sourceSequence(), record.changeSetId(),
+                            record.modelHash(), record.payloadSha256(), record.payloadJson()));
+                    return new InboxOutcome(outcome.duplicate(), outcome.checkpoint());
+                });
+        assertEquals(2, duckPublished.acknowledged());
+        assertEquals(0, duckPublished.status().backlog());
+        var duckOrder = duckSql.queryForList("SELECT version, name, memo FROM order_data WHERE id = ?",
+                new Object[] {42L}).get(0);
+        assertEquals(2L, ((Number) duckOrder.get("version")).longValue());
+        assertEquals("submitted", duckOrder.get("name"));
+        assertEquals("set by database", duckOrder.get("memo"));
+        assertEquals(2, duckSql.queryForList("SELECT * FROM teaql_event_applied_payload",
+                new Object[0]).size());
+
+        sql.execute("DROP TABLE teaql_event_outbox");
+        order.updateProperty("name", "should roll back");
+        assertThrows(RuntimeException.class,
+                () -> order.auditAs("reject when outbox unavailable").save(context));
+        assertEquals("submitted", sql.queryForList("SELECT name FROM order_data WHERE id = ?",
+                new Object[] {42L}).get(0).get("name"));
+        assertEquals(2L, ((Number) sql.queryForList("SELECT version FROM order_data WHERE id = ?",
+                new Object[] {42L}).get(0).get("version")).longValue());
+
+        // Even if application code catches the inner save failure, the shared
+        // JDBC transaction must not commit a business row without its outbox.
+        order.updateProperty("name", "caught failure must roll back");
+        assertThrows(IllegalStateException.class, () -> sql.executeInTransaction(() -> {
+            try {
+                order.auditAs("reject caught outbox failure").save(context);
+            } catch (RuntimeException expected) {
+                // Simulate a caller that handles the failed save and returns.
+            }
+        }));
+        assertEquals("submitted", sql.queryForList("SELECT name FROM order_data WHERE id = ?",
+                new Object[] {42L}).get(0).get("name"));
+        assertEquals(2L, ((Number) sql.queryForList("SELECT version FROM order_data WHERE id = ?",
+                new Object[] {42L}).get(0).get("version")).longValue());
+        runtime.close();
+    }
+}

examples/committed-events/verify.sh

@@ -0,0 +1,27 @@
+#!/usr/bin/env bash
+set -euo pipefail
+repo="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
+cd "$repo"
+evidence="$(mktemp -d -t teaql-java-committed-generated.XXXXXXXX)"
+generated_library_hash() {
+  (cd "$repo/examples/standard-order" && rg --files generated/lib -g '!target/**' \
+    | LC_ALL=C sort | xargs -d '\n' sha256sum)
+}
+generated_library_hash > "$evidence/generated-before.sha256"
+for round in 1 2; do
+  mvn -B -q -pl examples/committed-events -am \
+    -Dtest=CommittedEventsExampleTest -Dsurefire.failIfNoSpecifiedTests=false test \
+    2>&1 | tee "$evidence/runtime-$round.log"
+  rg -Fq 'COMMITTED_DELIVERY_RECOVERY_PASS language=java ordered=2 lost_ack=1 restart=1 duplicate=1' \
+    "$evidence/runtime-$round.log"
+  echo "PASS Java committed-events example round $round"
+  mvn -B -q -Pruntime-examples -pl examples/standard-order -am \
+    -Dtest=StandardOrderCommittedTargetTest -Dsurefire.failIfNoSpecifiedTests=false test \
+    2>&1 | tee "$evidence/generated-$round.log"
+  rg -Fq 'JAVA_COMMITTED_GENERATED_Q_PASS source=sqlite target=duckdb outbox=1' \
+    "$evidence/generated-$round.log"
+  generated_library_hash > "$evidence/generated-after.sha256"
+  cmp "$evidence/generated-before.sha256" "$evidence/generated-after.sha256"
+  echo "PASS Java generated committed Q example round $round"
+done
+echo "Java generated committed Q evidence: $evidence"

examples/standard-order/pom.xml

@@ -10,7 +10,9 @@
     <dependency><groupId>com.example</groupId><artifactId>request-tree-order-service-core</artifactId><version>0.0.1-SNAPSHOT</version></dependency>
     <dependency><groupId>io.teaql</groupId><artifactId>teaql-runtime</artifactId></dependency>
     <dependency><groupId>io.teaql</groupId><artifactId>teaql-sqlite</artifactId></dependency>
+    <dependency><groupId>io.teaql</groupId><artifactId>teaql-duckdb</artifactId><version>${project.version}</version></dependency>
     <dependency><groupId>io.teaql</groupId><artifactId>teaql-provider-jdbc</artifactId></dependency>
+    <dependency><groupId>org.duckdb</groupId><artifactId>duckdb_jdbc</artifactId><version>1.5.5.1</version></dependency>
     <dependency><groupId>io.teaql</groupId><artifactId>teaql-query-json</artifactId></dependency>
     <dependency><groupId>io.teaql</groupId><artifactId>teaql-opentelemetry</artifactId><version>${project.version}</version><scope>test</scope></dependency>
     <dependency><groupId>junit</groupId><artifactId>junit</artifactId><scope>test</scope></dependency>