jamesnetherton commented on code in PR #9006: URL: https://github.com/apache/camel-quarkus/pull/9006#discussion_r3782811670
########## extensions-support/langchain4j/runtime/src/main/java/org/apache/camel/quarkus/component/support/langchain4j/tracker/jdbc/JdbcIngestionTracker.java: ########## @@ -0,0 +1,288 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.camel.quarkus.component.support.langchain4j.tracker.jdbc; + +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Timestamp; +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; + +import javax.sql.DataSource; + +import org.apache.camel.quarkus.component.support.langchain4j.tracker.IngestionTracker; + +/** + * JDBC-backed {@link IngestionTracker}. Deliberately dialect-free SQL (works on PostgreSQL and H2): + * upserts are update-then-insert rather than vendor MERGE/ON CONFLICT. + * + * <p> + * <strong>Concurrency assumption: at most one writer per {@code (pipeline, documentId)} at a + * time.</strong> The update-then-insert is not atomic (each statement auto-commits on its own, + * there is no transaction), so two concurrent writers for the same row can both see the + * {@code UPDATE} affect zero rows and then both attempt the {@code INSERT}, and one fails on the + * {@code (pipeline, doc_id)} primary key. Wrapping the two statements in a transaction would not + * remove this race without either {@code SERIALIZABLE} isolation or a dialect-specific upsert, + * both at odds with staying dialect-free; callers (the sync pass runner processes one document + * at a time per pipeline) are relied upon to hold this invariant instead. + * + * <p> + * <strong>Status: Experimental.</strong> See {@link IngestionTracker}. + */ +public class JdbcIngestionTracker implements IngestionTracker { + + static final String TABLE = "cq_ingestion_tracker"; + + private final DataSource dataSource; + + public JdbcIngestionTracker(DataSource dataSource) { + this.dataSource = dataSource; + } + + @Override + public void ensureSchema() { + String ddl = "CREATE TABLE IF NOT EXISTS " + TABLE + " (" + + "pipeline VARCHAR(128) NOT NULL, " + + "doc_id VARCHAR(512) NOT NULL, " + + "fingerprint VARCHAR(256), " + + "content_hash VARCHAR(64), " + + "segment_count INT DEFAULT 0 NOT NULL, " + + "intended_count INT DEFAULT 0 NOT NULL, " + + "status VARCHAR(16) NOT NULL, " + + "origin VARCHAR(16) DEFAULT 'source' NOT NULL, " + + "tombstone BOOLEAN DEFAULT FALSE NOT NULL, " + + "pinned BOOLEAN DEFAULT FALSE NOT NULL, " + + "updated_at TIMESTAMP, " + + "PRIMARY KEY (pipeline, doc_id))"; + try (Connection connection = dataSource.getConnection()) { + connection.createStatement().execute(ddl); + } catch (SQLException e) { + throw new IllegalStateException( + "Cannot create the ingestion tracker table '" + TABLE + "'. " + + "mode=sync needs a working datasource — configure quarkus.datasource " + + "(Dev Services provides one in dev mode) or use mode=append.", + e); + } + } + + private static final String ROW_COLUMNS = "fingerprint, content_hash, segment_count, intended_count, status, " + + "origin, tombstone, pinned"; + + @Override + public Optional<TrackerRow> read(String pipeline, String documentId) { + String sql = "SELECT " + ROW_COLUMNS + " FROM " + TABLE + " WHERE pipeline = ? AND doc_id = ?"; + try (Connection connection = dataSource.getConnection(); + PreparedStatement statement = connection.prepareStatement(sql)) { + statement.setString(1, pipeline); + statement.setString(2, documentId); + try (ResultSet resultSet = statement.executeQuery()) { + if (!resultSet.next()) { + return Optional.empty(); + } + return Optional.of(row(pipeline, documentId, resultSet)); + } + } catch (SQLException e) { + throw new IllegalStateException("Tracker read failed for '" + documentId + "'", e); + } + } + + @Override + public List<TrackerRow> listDocuments(String pipeline) { + String sql = "SELECT doc_id, " + ROW_COLUMNS + " FROM " + TABLE + " WHERE pipeline = ?"; + try (Connection connection = dataSource.getConnection(); + PreparedStatement statement = connection.prepareStatement(sql)) { + statement.setString(1, pipeline); + try (ResultSet resultSet = statement.executeQuery()) { + List<TrackerRow> rows = new ArrayList<>(); + while (resultSet.next()) { + String documentId = resultSet.getString(1); + rows.add(new TrackerRow(pipeline, documentId, + resultSet.getString(2), resultSet.getString(3), + resultSet.getInt(4), resultSet.getInt(5), resultSet.getString(6), + resultSet.getString(7), resultSet.getBoolean(8), resultSet.getBoolean(9))); + } + return rows; + } + } catch (SQLException e) { + throw new IllegalStateException("Tracker listing failed for pipeline '" + pipeline + "'", e); + } + } + + private static TrackerRow row(String pipeline, String documentId, ResultSet resultSet) throws SQLException { + return new TrackerRow(pipeline, documentId, + resultSet.getString(1), resultSet.getString(2), + resultSet.getInt(3), resultSet.getInt(4), resultSet.getString(5), + resultSet.getString(6), resultSet.getBoolean(7), resultSet.getBoolean(8)); + } + + @Override + public void writeIntent(String pipeline, String documentId, String fingerprint, String contentHash, + int committedCount, int intendedCount, String origin) { + try (Connection connection = dataSource.getConnection()) { + String update = "UPDATE " + TABLE + " SET fingerprint = ?, content_hash = ?, " + + "intended_count = ?, status = 'in_progress', origin = ?, updated_at = ? " Review Comment: `intended_count = ?` overwrites unconditionally, so the shrink bound documented in the README ("the shrink bound `max(committed, intended)` rides along in the row") isn't actually held across an uncommitted intent — see the reproduction in the review body. Once an intent for N is durable, up to N segments may exist in the store, so the bound has to stay >= N until a `commit` proves the sweep happened. `commit` lowering it is correct; this path lowering it isn't. Still dialect-free: ```sql intended_count = CASE WHEN intended_count > ? THEN intended_count ELSE ? END ``` ########## extensions-support/langchain4j/runtime/src/main/java/org/apache/camel/quarkus/component/support/langchain4j/tracker/IngestionTracker.java: ########## @@ -0,0 +1,128 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.camel.quarkus.component.support.langchain4j.tracker; + +import java.util.List; +import java.util.Optional; + +/** + * The ingestion tracker: one row per document, the authority on what was ingested. The vector store is + * a projection that is never asked questions — losing the tracker costs re-ingestion (which + * converges thanks to deterministic segment ids), never correctness. Reconciliation is + * tracker-versus-source; the store is never enumerated. + * + * <p> + * <strong>Status: Experimental.</strong> Internal SPI — not a public API. No compatibility + * guarantees between releases; applications must not implement or call this interface, and it + * may change or be removed without a deprecation cycle. It exists so the implementation can be + * replaced wholesale: LangChain4j has no equivalent of LangChain-Python's {@code RecordManager} yet + * (<a href="https://github.com/langchain4j/langchain4j/issues/2931";>langchain4j#2931</a>), and + * once one lands upstream, an adapter implementing this interface replaces the + * {@code tracker.jdbc} package while every consumer stays untouched. New implementations must + * pass the behavioural tests of {@code JdbcIngestionTrackerTest}, whose test methods are written + * against this SPI only and are meant to be extracted into a shared contract base class the day + * a second implementation exists. + * + * <p> + * Two method groups, by replaceability: + * <ul> + * <li><em>Tracker subset</em> — {@link #ensureSchema}, {@link #read}, {@link #listDocuments}, + * {@link #refreshFingerprint}, {@link #deleteRow}: mirrors what an upstream record manager + * provides and would delegate to it directly.</li> + * <li><em>Camel Quarkus extensions</em> — the two-phase {@link #writeIntent}/{@link #commit} + * protocol, {@link #tombstone}/{@link #unsuppress}, {@link #pin}/{@link #unpin}, + * {@link #markFailed}: product semantics an upstream tracker will not carry; an adapter keeps + * these in side storage keyed the same way.</li> + * </ul> + */ +public interface IngestionTracker { + + String ORIGIN_SOURCE = "source"; + String ORIGIN_API = "api"; + + /** Creates or migrates the backing schema. Called once before first use. */ + void ensureSchema(); + + Optional<TrackerRow> read(String pipeline, String documentId); + + /** All rows of a pipeline — the reconciliation input. */ + List<TrackerRow> listDocuments(String pipeline); + + /** + * Durably records the intent to (re)write a document <em>before</em> the store is touched. + * A row left {@code in_progress} by a crash is never skipped by change detection, so the + * next delivery of the document converges the store. + */ + void writeIntent(String pipeline, String documentId, String fingerprint, String contentHash, + int committedCount, int intendedCount, String origin); Review Comment: `committedCount` is only read on the INSERT branch of the JDBC impl — passing `row.maxKnownCount()` back in on a re-intent has no effect at all (I checked). As it stands it's a parameter that looks like it preserves the bound and doesn't. Either honour it on the update path or drop it from the signature. ########## extensions-support/langchain4j/runtime/src/main/java/org/apache/camel/quarkus/component/support/langchain4j/tracker/jdbc/JdbcIngestionTracker.java: ########## @@ -0,0 +1,288 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.camel.quarkus.component.support.langchain4j.tracker.jdbc; + +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Timestamp; +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; + +import javax.sql.DataSource; + +import org.apache.camel.quarkus.component.support.langchain4j.tracker.IngestionTracker; + +/** + * JDBC-backed {@link IngestionTracker}. Deliberately dialect-free SQL (works on PostgreSQL and H2): + * upserts are update-then-insert rather than vendor MERGE/ON CONFLICT. + * + * <p> + * <strong>Concurrency assumption: at most one writer per {@code (pipeline, documentId)} at a + * time.</strong> The update-then-insert is not atomic (each statement auto-commits on its own, + * there is no transaction), so two concurrent writers for the same row can both see the + * {@code UPDATE} affect zero rows and then both attempt the {@code INSERT}, and one fails on the + * {@code (pipeline, doc_id)} primary key. Wrapping the two statements in a transaction would not + * remove this race without either {@code SERIALIZABLE} isolation or a dialect-specific upsert, + * both at odds with staying dialect-free; callers (the sync pass runner processes one document + * at a time per pipeline) are relied upon to hold this invariant instead. + * + * <p> + * <strong>Status: Experimental.</strong> See {@link IngestionTracker}. + */ +public class JdbcIngestionTracker implements IngestionTracker { + + static final String TABLE = "cq_ingestion_tracker"; Review Comment: Everything caller-supplied goes through `PreparedStatement` parameters and `TABLE` is a constant, so there's no injection today. But a `sync` mode will almost certainly grow a configurable table/schema name, and this concatenation becomes injectable the moment it does. Worth a comment here saying it must never become caller-supplied without identifier validation. ########## extensions-support/langchain4j/runtime/src/main/java/org/apache/camel/quarkus/component/support/langchain4j/tracker/jdbc/JdbcIngestionTracker.java: ########## @@ -0,0 +1,288 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.camel.quarkus.component.support.langchain4j.tracker.jdbc; + +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Timestamp; +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; + +import javax.sql.DataSource; + +import org.apache.camel.quarkus.component.support.langchain4j.tracker.IngestionTracker; + +/** + * JDBC-backed {@link IngestionTracker}. Deliberately dialect-free SQL (works on PostgreSQL and H2): + * upserts are update-then-insert rather than vendor MERGE/ON CONFLICT. + * + * <p> + * <strong>Concurrency assumption: at most one writer per {@code (pipeline, documentId)} at a + * time.</strong> The update-then-insert is not atomic (each statement auto-commits on its own, + * there is no transaction), so two concurrent writers for the same row can both see the + * {@code UPDATE} affect zero rows and then both attempt the {@code INSERT}, and one fails on the + * {@code (pipeline, doc_id)} primary key. Wrapping the two statements in a transaction would not + * remove this race without either {@code SERIALIZABLE} isolation or a dialect-specific upsert, + * both at odds with staying dialect-free; callers (the sync pass runner processes one document + * at a time per pipeline) are relied upon to hold this invariant instead. + * + * <p> + * <strong>Status: Experimental.</strong> See {@link IngestionTracker}. + */ +public class JdbcIngestionTracker implements IngestionTracker { + + static final String TABLE = "cq_ingestion_tracker"; + + private final DataSource dataSource; + + public JdbcIngestionTracker(DataSource dataSource) { + this.dataSource = dataSource; + } + + @Override + public void ensureSchema() { + String ddl = "CREATE TABLE IF NOT EXISTS " + TABLE + " (" + + "pipeline VARCHAR(128) NOT NULL, " + + "doc_id VARCHAR(512) NOT NULL, " + + "fingerprint VARCHAR(256), " + + "content_hash VARCHAR(64), " + + "segment_count INT DEFAULT 0 NOT NULL, " + + "intended_count INT DEFAULT 0 NOT NULL, " + + "status VARCHAR(16) NOT NULL, " + + "origin VARCHAR(16) DEFAULT 'source' NOT NULL, " + + "tombstone BOOLEAN DEFAULT FALSE NOT NULL, " + + "pinned BOOLEAN DEFAULT FALSE NOT NULL, " + + "updated_at TIMESTAMP, " + + "PRIMARY KEY (pipeline, doc_id))"; + try (Connection connection = dataSource.getConnection()) { + connection.createStatement().execute(ddl); + } catch (SQLException e) { + throw new IllegalStateException( + "Cannot create the ingestion tracker table '" + TABLE + "'. " + + "mode=sync needs a working datasource — configure quarkus.datasource " + + "(Dev Services provides one in dev mode) or use mode=append.", + e); + } + } + + private static final String ROW_COLUMNS = "fingerprint, content_hash, segment_count, intended_count, status, " + + "origin, tombstone, pinned"; + + @Override + public Optional<TrackerRow> read(String pipeline, String documentId) { + String sql = "SELECT " + ROW_COLUMNS + " FROM " + TABLE + " WHERE pipeline = ? AND doc_id = ?"; + try (Connection connection = dataSource.getConnection(); + PreparedStatement statement = connection.prepareStatement(sql)) { + statement.setString(1, pipeline); + statement.setString(2, documentId); + try (ResultSet resultSet = statement.executeQuery()) { + if (!resultSet.next()) { + return Optional.empty(); + } + return Optional.of(row(pipeline, documentId, resultSet)); + } + } catch (SQLException e) { + throw new IllegalStateException("Tracker read failed for '" + documentId + "'", e); + } + } + + @Override + public List<TrackerRow> listDocuments(String pipeline) { + String sql = "SELECT doc_id, " + ROW_COLUMNS + " FROM " + TABLE + " WHERE pipeline = ?"; + try (Connection connection = dataSource.getConnection(); + PreparedStatement statement = connection.prepareStatement(sql)) { + statement.setString(1, pipeline); + try (ResultSet resultSet = statement.executeQuery()) { + List<TrackerRow> rows = new ArrayList<>(); + while (resultSet.next()) { + String documentId = resultSet.getString(1); + rows.add(new TrackerRow(pipeline, documentId, + resultSet.getString(2), resultSet.getString(3), + resultSet.getInt(4), resultSet.getInt(5), resultSet.getString(6), + resultSet.getString(7), resultSet.getBoolean(8), resultSet.getBoolean(9))); + } + return rows; + } + } catch (SQLException e) { + throw new IllegalStateException("Tracker listing failed for pipeline '" + pipeline + "'", e); + } + } + + private static TrackerRow row(String pipeline, String documentId, ResultSet resultSet) throws SQLException { + return new TrackerRow(pipeline, documentId, + resultSet.getString(1), resultSet.getString(2), + resultSet.getInt(3), resultSet.getInt(4), resultSet.getString(5), + resultSet.getString(6), resultSet.getBoolean(7), resultSet.getBoolean(8)); + } + + @Override + public void writeIntent(String pipeline, String documentId, String fingerprint, String contentHash, + int committedCount, int intendedCount, String origin) { + try (Connection connection = dataSource.getConnection()) { + String update = "UPDATE " + TABLE + " SET fingerprint = ?, content_hash = ?, " + + "intended_count = ?, status = 'in_progress', origin = ?, updated_at = ? " + + "WHERE pipeline = ? AND doc_id = ?"; + int updated; + try (PreparedStatement statement = connection.prepareStatement(update)) { + statement.setString(1, fingerprint); + statement.setString(2, contentHash); + statement.setInt(3, intendedCount); + statement.setString(4, origin); + statement.setTimestamp(5, Timestamp.from(Instant.now())); + statement.setString(6, pipeline); + statement.setString(7, documentId); + updated = statement.executeUpdate(); + } + if (updated == 0) { + String insert = "INSERT INTO " + TABLE + + " (pipeline, doc_id, fingerprint, content_hash, segment_count, intended_count, " + + "status, origin, updated_at) VALUES (?, ?, ?, ?, ?, ?, 'in_progress', ?, ?)"; + try (PreparedStatement statement = connection.prepareStatement(insert)) { + statement.setString(1, pipeline); + statement.setString(2, documentId); + statement.setString(3, fingerprint); + statement.setString(4, contentHash); + statement.setInt(5, committedCount); + statement.setInt(6, intendedCount); + statement.setString(7, origin); + statement.setTimestamp(8, Timestamp.from(Instant.now())); + statement.executeUpdate(); + } + } + } catch (SQLException e) { + throw new IllegalStateException("Tracker intent write failed for '" + documentId + "'", e); + } + } + + @Override + public void tombstone(String pipeline, String documentId) { + setFlag(pipeline, documentId, "tombstone", true); + } + + @Override + public void unsuppress(String pipeline, String documentId) { + setFlag(pipeline, documentId, "tombstone", false); + } + + @Override + public void pin(String pipeline, String documentId) { + setFlag(pipeline, documentId, "pinned", true); + } + + @Override + public void unpin(String pipeline, String documentId) { + setFlag(pipeline, documentId, "pinned", false); + } + + private void setFlag(String pipeline, String documentId, String column, boolean value) { Review Comment: `setFlag` is UPDATE-only, so `tombstone`/`pin`/`unsuppress`/`unpin` silently no-op when no row exists — `tombstone("p", "ghost")` leaves no row and throws nothing. The README (line 80) explicitly promises the opposite: "a tombstone can even be recorded for a never-ingested document, as a pure suppression record". So "deleted stays deleted" doesn't hold for a document deleted before its first successful ingest — the next pass re-ingests it. Either upsert a suppression row here or correct the README. ########## extensions-support/langchain4j/runtime/src/test/java/org/apache/camel/quarkus/component/support/langchain4j/tracker/jdbc/JdbcIngestionTrackerTest.java: ########## @@ -0,0 +1,169 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.camel.quarkus.component.support.langchain4j.tracker.jdbc; + +import java.util.List; +import java.util.UUID; + +import org.apache.camel.quarkus.component.support.langchain4j.tracker.IngestionTracker; +import org.h2.jdbcx.JdbcDataSource; +import org.junit.jupiter.api.BeforeEach; +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.assertTrue; + +/** + * The behavioural contract of the {@link IngestionTracker} SPI, exercised against the JDBC + * implementation on H2. The test methods are deliberately written against the SPI only — nothing + * below {@link #setUpTracker()} references {@link JdbcIngestionTracker}. + * + * <p> + * The contract lives folded into this class only because a single implementation exists today. + * When a second one arrives — e.g. an adapter over a future upstream LangChain4j record manager + * (langchain4j#2931) — extract the test methods into an abstract {@code IngestionTrackerContract} + * base class with a {@code createTracker()} factory, and keep one {@code *Test} subclass per + * implementation: a replacement is a drop-in exactly when its subclass passes. + */ +class JdbcIngestionTrackerTest { + + IngestionTracker tracker; + + /** A fresh, empty tracker per test. */ + @BeforeEach + void setUpTracker() { + JdbcDataSource dataSource = new JdbcDataSource(); + dataSource.setURL("jdbc:h2:mem:" + UUID.randomUUID() + ";DB_CLOSE_DELAY=-1"); + tracker = new JdbcIngestionTracker(dataSource); + tracker.ensureSchema(); + } + + @Test + void unknownDocumentReadsEmpty() { + assertTrue(tracker.read("p", "missing").isEmpty()); + } + + @Test + void intentIsDurableAndNeverSkippable() { + tracker.writeIntent("p", "doc", "fp1", "hash1", 0, 5, IngestionTracker.ORIGIN_SOURCE); + + IngestionTracker.TrackerRow row = tracker.read("p", "doc").orElseThrow(); + assertFalse(row.done(), "an intent row must not count as done"); + assertEquals(5, row.maxKnownCount(), "the shrink bound must cover the intended count"); + } + + @Test + void commitCompletesTheIntent() { + tracker.writeIntent("p", "doc", "fp1", "hash1", 0, 5, IngestionTracker.ORIGIN_SOURCE); + tracker.commit("p", "doc", "fp1", "hash1", 3); + + IngestionTracker.TrackerRow row = tracker.read("p", "doc").orElseThrow(); + assertTrue(row.done()); + assertEquals("fp1", row.fingerprint()); + assertEquals("hash1", row.contentHash()); + assertEquals(3, row.segmentCount()); + } + + @Test + void reintentKeepsTheLargestKnownCount() { Review Comment: This only covers re-intent *after* a commit — the path where `segment_count` carries the bound, which works. The intent -> intent path is the broken one and isn't covered. Also missing: `markFailed` straight from `in_progress`, `commit` without a preceding intent, and `refreshFingerprint` on a non-`done` row. ########## extensions-support/langchain4j/runtime/src/main/java/org/apache/camel/quarkus/component/support/langchain4j/tracker/jdbc/JdbcIngestionTracker.java: ########## @@ -0,0 +1,288 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.camel.quarkus.component.support.langchain4j.tracker.jdbc; + +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Timestamp; +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; + +import javax.sql.DataSource; + +import org.apache.camel.quarkus.component.support.langchain4j.tracker.IngestionTracker; + +/** + * JDBC-backed {@link IngestionTracker}. Deliberately dialect-free SQL (works on PostgreSQL and H2): + * upserts are update-then-insert rather than vendor MERGE/ON CONFLICT. + * + * <p> + * <strong>Concurrency assumption: at most one writer per {@code (pipeline, documentId)} at a + * time.</strong> The update-then-insert is not atomic (each statement auto-commits on its own, + * there is no transaction), so two concurrent writers for the same row can both see the + * {@code UPDATE} affect zero rows and then both attempt the {@code INSERT}, and one fails on the + * {@code (pipeline, doc_id)} primary key. Wrapping the two statements in a transaction would not + * remove this race without either {@code SERIALIZABLE} isolation or a dialect-specific upsert, + * both at odds with staying dialect-free; callers (the sync pass runner processes one document + * at a time per pipeline) are relied upon to hold this invariant instead. + * + * <p> + * <strong>Status: Experimental.</strong> See {@link IngestionTracker}. + */ +public class JdbcIngestionTracker implements IngestionTracker { + + static final String TABLE = "cq_ingestion_tracker"; + + private final DataSource dataSource; + + public JdbcIngestionTracker(DataSource dataSource) { + this.dataSource = dataSource; + } + + @Override + public void ensureSchema() { + String ddl = "CREATE TABLE IF NOT EXISTS " + TABLE + " (" + + "pipeline VARCHAR(128) NOT NULL, " + + "doc_id VARCHAR(512) NOT NULL, " Review Comment: `VARCHAR(512)` hard-fails rather than truncating on longer keys (confirmed: `JdbcSQLDataException`). S3 keys go to 1024 bytes and file URIs can exceed 512 too. ########## extensions-support/langchain4j/README.adoc: ########## @@ -0,0 +1,107 @@ += LangChain4j Support + +This module provides common support code shared by Camel Quarkus LangChain4j extensions. + +== Ingestion tracker + +Status: *Experimental*. + +`org.apache.camel.quarkus.component.support.langchain4j.tracker.IngestionTracker` is the bookkeeping +SPI behind keeping a vector store in sync with a changing document source (files, buckets, feeds, +...). Vector stores cannot be enumerated for "what did I already ingest", and LangChain4j has no +equivalent of LangChain-Python's `RecordManager` +(https://github.com/langchain4j/langchain4j/issues/2931[langchain4j#2931]), so without external +bookkeeping any ingestion pipeline is either append-only (restarts re-embed the whole corpus, +edited documents join their previous vectors instead of replacing them) or wipe-and-reload +(loses everything between passes). `IngestionTracker` closes that gap: it tracks one row per document +— fingerprint, content hash, segment count, a two-phase `in_progress`/`done`/`failed` status, +and `tombstone`/`pinned` flags — as the single authority on what was ingested. The vector store +itself stays a disposable projection that is never asked questions; losing the tracker only costs +re-ingestion, never correctness, because segment ids are deterministic. + +The two-phase `writeIntent`/`commit` protocol is what makes the tracker crash-safe: a row left +`in_progress` by a crash (killed process, OOM, ...) is never mistaken for "already ingested", so +the next delivery of the same document converges the store instead of skipping it. + +=== Document states + +Each document is one row that moves through the states below. Transitions are the SPI methods; +the annotations in parentheses describe when the consumer (the ingestion pipeline) invokes them. + +---- + (no row) + │ writeIntent durable BEFORE the store is touched; a crash strands + ▼ the row here, and an in_progress row is never trusted + in_progress as "already ingested" — the next delivery of the + │ commit document converges the store + ▼ + done ◄──────────────────────┐ + │ │ + ├─ (unchanged) ──► done │ no transition: the skip that makes restarts free + │ │ + ├─ refreshFingerprint ────┘ fingerprint renewed after a content-hash match + │ (tier-2), so the cheap tier-1 check works next pass + │ + ├─ writeIntent ──► in_progress (edited content: replace; the shrink bound + │ max(committed, intended) rides along in the row) + │ + ├─ markFailed ──► failed (processing failure — also fired straight from + │ │ in_progress; previously committed segments, if + │ │ any — a stale but valid version — keep serving) + │ │ + │ ├─ (fingerprint unchanged) ──► failed dead-lettered: skipped on every + │ │ pass until the content changes + │ │ + │ └─ writeIntent ──► in_progress (changed content: retry) + │ + ├─ tombstone() ──► done(0) + tombstone (explicit delete: the consumer removes + │ │ the vectors and records commit(0) with + │ │ the flag) — "deleted stays deleted": + │ │ every future ingest of the document is + │ │ suppressed, even though the source + │ │ still has it + │ │ + │ └─ unsuppress() ──► done(0) suppression lifted: the next pass + │ re-ingests the document + │ + └─ pin() ──► done + pinned (an API write corrected a source-owned document) + │ — "the correction wins": source-origin updates + │ are suppressed; API-origin writes still pass + │ + └─ unpin() ──► done the source owns the document again + + any state ── deleteRow ──► (no row) reconciliation: the source no longer lists the + document (guarded by the consumer's pass + interlock, never fired by the tracker itself) +---- + +`tombstone` and `pinned` are columns orthogonal to `status`: drawn above off `done` because +that is how the consumer composes them, but they can accompany any state (a tombstone can even +be recorded for a never-ingested document, as a pure suppression record) and they survive every Review Comment: This claim isn't implemented — see the comment on `setFlag`. While you're here: worth an operator note that this table is a durable inventory of the corpus. Content correctly never lands in it, but `doc_id` is typically a path, URL or S3 key, and those routinely encode sensitive structure (`/customers/12345/contract.pdf`). Doc ids also go into every exception message, so they reach logs and stack traces. ########## extensions-support/langchain4j/runtime/src/main/java/org/apache/camel/quarkus/component/support/langchain4j/tracker/jdbc/JdbcIngestionTracker.java: ########## @@ -0,0 +1,288 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.camel.quarkus.component.support.langchain4j.tracker.jdbc; + +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Timestamp; +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; + +import javax.sql.DataSource; + +import org.apache.camel.quarkus.component.support.langchain4j.tracker.IngestionTracker; + +/** + * JDBC-backed {@link IngestionTracker}. Deliberately dialect-free SQL (works on PostgreSQL and H2): + * upserts are update-then-insert rather than vendor MERGE/ON CONFLICT. + * + * <p> + * <strong>Concurrency assumption: at most one writer per {@code (pipeline, documentId)} at a + * time.</strong> The update-then-insert is not atomic (each statement auto-commits on its own, + * there is no transaction), so two concurrent writers for the same row can both see the + * {@code UPDATE} affect zero rows and then both attempt the {@code INSERT}, and one fails on the + * {@code (pipeline, doc_id)} primary key. Wrapping the two statements in a transaction would not + * remove this race without either {@code SERIALIZABLE} isolation or a dialect-specific upsert, + * both at odds with staying dialect-free; callers (the sync pass runner processes one document + * at a time per pipeline) are relied upon to hold this invariant instead. + * + * <p> + * <strong>Status: Experimental.</strong> See {@link IngestionTracker}. + */ +public class JdbcIngestionTracker implements IngestionTracker { + + static final String TABLE = "cq_ingestion_tracker"; + + private final DataSource dataSource; + + public JdbcIngestionTracker(DataSource dataSource) { + this.dataSource = dataSource; + } + + @Override + public void ensureSchema() { + String ddl = "CREATE TABLE IF NOT EXISTS " + TABLE + " (" + + "pipeline VARCHAR(128) NOT NULL, " + + "doc_id VARCHAR(512) NOT NULL, " + + "fingerprint VARCHAR(256), " + + "content_hash VARCHAR(64), " + + "segment_count INT DEFAULT 0 NOT NULL, " + + "intended_count INT DEFAULT 0 NOT NULL, " + + "status VARCHAR(16) NOT NULL, " + + "origin VARCHAR(16) DEFAULT 'source' NOT NULL, " + + "tombstone BOOLEAN DEFAULT FALSE NOT NULL, " + + "pinned BOOLEAN DEFAULT FALSE NOT NULL, " + + "updated_at TIMESTAMP, " + + "PRIMARY KEY (pipeline, doc_id))"; + try (Connection connection = dataSource.getConnection()) { + connection.createStatement().execute(ddl); + } catch (SQLException e) { + throw new IllegalStateException( + "Cannot create the ingestion tracker table '" + TABLE + "'. " + + "mode=sync needs a working datasource — configure quarkus.datasource " + + "(Dev Services provides one in dev mode) or use mode=append.", + e); + } + } + + private static final String ROW_COLUMNS = "fingerprint, content_hash, segment_count, intended_count, status, " + + "origin, tombstone, pinned"; + + @Override + public Optional<TrackerRow> read(String pipeline, String documentId) { + String sql = "SELECT " + ROW_COLUMNS + " FROM " + TABLE + " WHERE pipeline = ? AND doc_id = ?"; + try (Connection connection = dataSource.getConnection(); + PreparedStatement statement = connection.prepareStatement(sql)) { + statement.setString(1, pipeline); + statement.setString(2, documentId); + try (ResultSet resultSet = statement.executeQuery()) { + if (!resultSet.next()) { + return Optional.empty(); + } + return Optional.of(row(pipeline, documentId, resultSet)); + } + } catch (SQLException e) { + throw new IllegalStateException("Tracker read failed for '" + documentId + "'", e); + } + } + + @Override + public List<TrackerRow> listDocuments(String pipeline) { + String sql = "SELECT doc_id, " + ROW_COLUMNS + " FROM " + TABLE + " WHERE pipeline = ?"; + try (Connection connection = dataSource.getConnection(); + PreparedStatement statement = connection.prepareStatement(sql)) { + statement.setString(1, pipeline); + try (ResultSet resultSet = statement.executeQuery()) { + List<TrackerRow> rows = new ArrayList<>(); + while (resultSet.next()) { + String documentId = resultSet.getString(1); + rows.add(new TrackerRow(pipeline, documentId, + resultSet.getString(2), resultSet.getString(3), + resultSet.getInt(4), resultSet.getInt(5), resultSet.getString(6), + resultSet.getString(7), resultSet.getBoolean(8), resultSet.getBoolean(9))); + } + return rows; + } + } catch (SQLException e) { + throw new IllegalStateException("Tracker listing failed for pipeline '" + pipeline + "'", e); + } + } + + private static TrackerRow row(String pipeline, String documentId, ResultSet resultSet) throws SQLException { + return new TrackerRow(pipeline, documentId, + resultSet.getString(1), resultSet.getString(2), + resultSet.getInt(3), resultSet.getInt(4), resultSet.getString(5), + resultSet.getString(6), resultSet.getBoolean(7), resultSet.getBoolean(8)); + } + + @Override + public void writeIntent(String pipeline, String documentId, String fingerprint, String contentHash, + int committedCount, int intendedCount, String origin) { + try (Connection connection = dataSource.getConnection()) { + String update = "UPDATE " + TABLE + " SET fingerprint = ?, content_hash = ?, " + + "intended_count = ?, status = 'in_progress', origin = ?, updated_at = ? " + + "WHERE pipeline = ? AND doc_id = ?"; + int updated; + try (PreparedStatement statement = connection.prepareStatement(update)) { + statement.setString(1, fingerprint); + statement.setString(2, contentHash); + statement.setInt(3, intendedCount); + statement.setString(4, origin); + statement.setTimestamp(5, Timestamp.from(Instant.now())); + statement.setString(6, pipeline); + statement.setString(7, documentId); + updated = statement.executeUpdate(); + } + if (updated == 0) { + String insert = "INSERT INTO " + TABLE + + " (pipeline, doc_id, fingerprint, content_hash, segment_count, intended_count, " + + "status, origin, updated_at) VALUES (?, ?, ?, ?, ?, ?, 'in_progress', ?, ?)"; + try (PreparedStatement statement = connection.prepareStatement(insert)) { + statement.setString(1, pipeline); + statement.setString(2, documentId); + statement.setString(3, fingerprint); + statement.setString(4, contentHash); + statement.setInt(5, committedCount); + statement.setInt(6, intendedCount); + statement.setString(7, origin); + statement.setTimestamp(8, Timestamp.from(Instant.now())); + statement.executeUpdate(); + } + } + } catch (SQLException e) { + throw new IllegalStateException("Tracker intent write failed for '" + documentId + "'", e); + } + } + + @Override + public void tombstone(String pipeline, String documentId) { + setFlag(pipeline, documentId, "tombstone", true); + } + + @Override + public void unsuppress(String pipeline, String documentId) { + setFlag(pipeline, documentId, "tombstone", false); + } + + @Override + public void pin(String pipeline, String documentId) { + setFlag(pipeline, documentId, "pinned", true); + } + + @Override + public void unpin(String pipeline, String documentId) { + setFlag(pipeline, documentId, "pinned", false); + } + + private void setFlag(String pipeline, String documentId, String column, boolean value) { + String sql = "UPDATE " + TABLE + " SET " + column + " = ?, updated_at = ? " + + "WHERE pipeline = ? AND doc_id = ?"; + try (Connection connection = dataSource.getConnection(); + PreparedStatement statement = connection.prepareStatement(sql)) { + statement.setBoolean(1, value); + statement.setTimestamp(2, Timestamp.from(Instant.now())); + statement.setString(3, pipeline); + statement.setString(4, documentId); + statement.executeUpdate(); + } catch (SQLException e) { + throw new IllegalStateException("Tracker " + column + " update failed for '" + documentId + "'", e); + } + } + + @Override + public void markFailed(String pipeline, String documentId, String fingerprint) { + try (Connection connection = dataSource.getConnection()) { + String update = "UPDATE " + TABLE + " SET fingerprint = ?, status = 'failed', updated_at = ? " + + "WHERE pipeline = ? AND doc_id = ?"; + int updated; + try (PreparedStatement statement = connection.prepareStatement(update)) { + statement.setString(1, fingerprint); + statement.setTimestamp(2, Timestamp.from(Instant.now())); + statement.setString(3, pipeline); + statement.setString(4, documentId); + updated = statement.executeUpdate(); + } + if (updated == 0) { + String insert = "INSERT INTO " + TABLE + + " (pipeline, doc_id, fingerprint, status, origin, updated_at) " + + "VALUES (?, ?, ?, 'failed', 'source', ?)"; Review Comment: The INSERT branch hardcodes `origin = 'source'`, so a document that only ever fails records the wrong origin. ########## integration-tests/langchain4j-ingestion-tracker/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingestiontracker/it/Langchain4jIngestionTrackerTest.java: ########## @@ -0,0 +1,163 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.camel.quarkus.component.langchain4j.ingestiontracker.it; + +import io.quarkus.test.junit.QuarkusTest; +import io.restassured.RestAssured; +import org.junit.jupiter.api.Test; + +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.hasSize; + +/** + * Runs the {@code IngestionTracker} behavioural guarantees (mirrored from {@code JdbcIngestionTrackerTest}) + * against a real PostgreSQL server, provisioned by Quarkus Dev Services, through + * {@link IngestionTrackerResource}. + */ +@QuarkusTest +class Langchain4jIngestionTrackerTest { Review Comment: Every method here shares `pipeline="p"` / `docId="doc"` with no cleanup and no DB reset. It passes today only because each one happens to rewrite the row before asserting — `deleteRowForgetsTheDocument` asserting `hasSize(0)` on pipeline `p` is one added test away from being order-dependent. Unique ids per test, or a truncate in `@BeforeEach`, would remove the trap. ########## extensions-support/langchain4j/runtime/src/main/java/org/apache/camel/quarkus/component/support/langchain4j/tracker/jdbc/JdbcIngestionTracker.java: ########## @@ -0,0 +1,288 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.camel.quarkus.component.support.langchain4j.tracker.jdbc; + +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Timestamp; +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; + +import javax.sql.DataSource; + +import org.apache.camel.quarkus.component.support.langchain4j.tracker.IngestionTracker; + +/** + * JDBC-backed {@link IngestionTracker}. Deliberately dialect-free SQL (works on PostgreSQL and H2): + * upserts are update-then-insert rather than vendor MERGE/ON CONFLICT. + * + * <p> + * <strong>Concurrency assumption: at most one writer per {@code (pipeline, documentId)} at a + * time.</strong> The update-then-insert is not atomic (each statement auto-commits on its own, + * there is no transaction), so two concurrent writers for the same row can both see the + * {@code UPDATE} affect zero rows and then both attempt the {@code INSERT}, and one fails on the + * {@code (pipeline, doc_id)} primary key. Wrapping the two statements in a transaction would not + * remove this race without either {@code SERIALIZABLE} isolation or a dialect-specific upsert, + * both at odds with staying dialect-free; callers (the sync pass runner processes one document + * at a time per pipeline) are relied upon to hold this invariant instead. + * + * <p> + * <strong>Status: Experimental.</strong> See {@link IngestionTracker}. + */ +public class JdbcIngestionTracker implements IngestionTracker { + + static final String TABLE = "cq_ingestion_tracker"; + + private final DataSource dataSource; + + public JdbcIngestionTracker(DataSource dataSource) { + this.dataSource = dataSource; + } + + @Override + public void ensureSchema() { + String ddl = "CREATE TABLE IF NOT EXISTS " + TABLE + " (" + + "pipeline VARCHAR(128) NOT NULL, " + + "doc_id VARCHAR(512) NOT NULL, " + + "fingerprint VARCHAR(256), " + + "content_hash VARCHAR(64), " + + "segment_count INT DEFAULT 0 NOT NULL, " + + "intended_count INT DEFAULT 0 NOT NULL, " + + "status VARCHAR(16) NOT NULL, " + + "origin VARCHAR(16) DEFAULT 'source' NOT NULL, " + + "tombstone BOOLEAN DEFAULT FALSE NOT NULL, " + + "pinned BOOLEAN DEFAULT FALSE NOT NULL, " + + "updated_at TIMESTAMP, " + + "PRIMARY KEY (pipeline, doc_id))"; + try (Connection connection = dataSource.getConnection()) { + connection.createStatement().execute(ddl); Review Comment: `CREATE TABLE IF NOT EXISTS` against the application's datasource forces the production DB user to hold DDL rights, with no opt-out. Worth mirroring `JdbcMessageIdRepository` in camel-sql: a `createTableIfNotExists` flag (default true) plus a `tableExists` probe, so operators can pre-provision and run least-privilege. When that becomes a config option it should be `ConfigPhase.RUN_TIME` per our security model doc. Also minor: `connection.createStatement()` isn't in the try-with-resources. ########## extensions-support/langchain4j/runtime/src/main/java/org/apache/camel/quarkus/component/support/langchain4j/tracker/jdbc/JdbcIngestionTracker.java: ########## @@ -0,0 +1,288 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.camel.quarkus.component.support.langchain4j.tracker.jdbc; + +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Timestamp; +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; + +import javax.sql.DataSource; + +import org.apache.camel.quarkus.component.support.langchain4j.tracker.IngestionTracker; + +/** + * JDBC-backed {@link IngestionTracker}. Deliberately dialect-free SQL (works on PostgreSQL and H2): + * upserts are update-then-insert rather than vendor MERGE/ON CONFLICT. + * + * <p> + * <strong>Concurrency assumption: at most one writer per {@code (pipeline, documentId)} at a + * time.</strong> The update-then-insert is not atomic (each statement auto-commits on its own, + * there is no transaction), so two concurrent writers for the same row can both see the + * {@code UPDATE} affect zero rows and then both attempt the {@code INSERT}, and one fails on the + * {@code (pipeline, doc_id)} primary key. Wrapping the two statements in a transaction would not + * remove this race without either {@code SERIALIZABLE} isolation or a dialect-specific upsert, + * both at odds with staying dialect-free; callers (the sync pass runner processes one document + * at a time per pipeline) are relied upon to hold this invariant instead. + * + * <p> + * <strong>Status: Experimental.</strong> See {@link IngestionTracker}. + */ +public class JdbcIngestionTracker implements IngestionTracker { + + static final String TABLE = "cq_ingestion_tracker"; + + private final DataSource dataSource; + + public JdbcIngestionTracker(DataSource dataSource) { + this.dataSource = dataSource; + } + + @Override + public void ensureSchema() { + String ddl = "CREATE TABLE IF NOT EXISTS " + TABLE + " (" + + "pipeline VARCHAR(128) NOT NULL, " + + "doc_id VARCHAR(512) NOT NULL, " + + "fingerprint VARCHAR(256), " + + "content_hash VARCHAR(64), " + + "segment_count INT DEFAULT 0 NOT NULL, " + + "intended_count INT DEFAULT 0 NOT NULL, " + + "status VARCHAR(16) NOT NULL, " + + "origin VARCHAR(16) DEFAULT 'source' NOT NULL, " + + "tombstone BOOLEAN DEFAULT FALSE NOT NULL, " + + "pinned BOOLEAN DEFAULT FALSE NOT NULL, " + + "updated_at TIMESTAMP, " + + "PRIMARY KEY (pipeline, doc_id))"; + try (Connection connection = dataSource.getConnection()) { + connection.createStatement().execute(ddl); + } catch (SQLException e) { + throw new IllegalStateException( + "Cannot create the ingestion tracker table '" + TABLE + "'. " + + "mode=sync needs a working datasource — configure quarkus.datasource " + + "(Dev Services provides one in dev mode) or use mode=append.", + e); + } + } + + private static final String ROW_COLUMNS = "fingerprint, content_hash, segment_count, intended_count, status, " + + "origin, tombstone, pinned"; + + @Override + public Optional<TrackerRow> read(String pipeline, String documentId) { + String sql = "SELECT " + ROW_COLUMNS + " FROM " + TABLE + " WHERE pipeline = ? AND doc_id = ?"; + try (Connection connection = dataSource.getConnection(); + PreparedStatement statement = connection.prepareStatement(sql)) { + statement.setString(1, pipeline); + statement.setString(2, documentId); + try (ResultSet resultSet = statement.executeQuery()) { + if (!resultSet.next()) { + return Optional.empty(); + } + return Optional.of(row(pipeline, documentId, resultSet)); + } + } catch (SQLException e) { + throw new IllegalStateException("Tracker read failed for '" + documentId + "'", e); + } + } + + @Override + public List<TrackerRow> listDocuments(String pipeline) { + String sql = "SELECT doc_id, " + ROW_COLUMNS + " FROM " + TABLE + " WHERE pipeline = ?"; + try (Connection connection = dataSource.getConnection(); + PreparedStatement statement = connection.prepareStatement(sql)) { + statement.setString(1, pipeline); + try (ResultSet resultSet = statement.executeQuery()) { + List<TrackerRow> rows = new ArrayList<>(); + while (resultSet.next()) { + String documentId = resultSet.getString(1); + rows.add(new TrackerRow(pipeline, documentId, + resultSet.getString(2), resultSet.getString(3), + resultSet.getInt(4), resultSet.getInt(5), resultSet.getString(6), + resultSet.getString(7), resultSet.getBoolean(8), resultSet.getBoolean(9))); + } + return rows; + } + } catch (SQLException e) { + throw new IllegalStateException("Tracker listing failed for pipeline '" + pipeline + "'", e); + } + } + + private static TrackerRow row(String pipeline, String documentId, ResultSet resultSet) throws SQLException { + return new TrackerRow(pipeline, documentId, + resultSet.getString(1), resultSet.getString(2), + resultSet.getInt(3), resultSet.getInt(4), resultSet.getString(5), + resultSet.getString(6), resultSet.getBoolean(7), resultSet.getBoolean(8)); + } + + @Override + public void writeIntent(String pipeline, String documentId, String fingerprint, String contentHash, + int committedCount, int intendedCount, String origin) { + try (Connection connection = dataSource.getConnection()) { + String update = "UPDATE " + TABLE + " SET fingerprint = ?, content_hash = ?, " + + "intended_count = ?, status = 'in_progress', origin = ?, updated_at = ? " + + "WHERE pipeline = ? AND doc_id = ?"; + int updated; + try (PreparedStatement statement = connection.prepareStatement(update)) { + statement.setString(1, fingerprint); + statement.setString(2, contentHash); + statement.setInt(3, intendedCount); + statement.setString(4, origin); + statement.setTimestamp(5, Timestamp.from(Instant.now())); + statement.setString(6, pipeline); + statement.setString(7, documentId); + updated = statement.executeUpdate(); + } + if (updated == 0) { + String insert = "INSERT INTO " + TABLE + + " (pipeline, doc_id, fingerprint, content_hash, segment_count, intended_count, " + + "status, origin, updated_at) VALUES (?, ?, ?, ?, ?, ?, 'in_progress', ?, ?)"; + try (PreparedStatement statement = connection.prepareStatement(insert)) { + statement.setString(1, pipeline); + statement.setString(2, documentId); + statement.setString(3, fingerprint); + statement.setString(4, contentHash); + statement.setInt(5, committedCount); + statement.setInt(6, intendedCount); + statement.setString(7, origin); + statement.setTimestamp(8, Timestamp.from(Instant.now())); + statement.executeUpdate(); + } + } + } catch (SQLException e) { + throw new IllegalStateException("Tracker intent write failed for '" + documentId + "'", e); + } + } + + @Override + public void tombstone(String pipeline, String documentId) { + setFlag(pipeline, documentId, "tombstone", true); + } + + @Override + public void unsuppress(String pipeline, String documentId) { + setFlag(pipeline, documentId, "tombstone", false); + } + + @Override + public void pin(String pipeline, String documentId) { + setFlag(pipeline, documentId, "pinned", true); + } + + @Override + public void unpin(String pipeline, String documentId) { + setFlag(pipeline, documentId, "pinned", false); + } + + private void setFlag(String pipeline, String documentId, String column, boolean value) { + String sql = "UPDATE " + TABLE + " SET " + column + " = ?, updated_at = ? " + + "WHERE pipeline = ? AND doc_id = ?"; + try (Connection connection = dataSource.getConnection(); + PreparedStatement statement = connection.prepareStatement(sql)) { + statement.setBoolean(1, value); + statement.setTimestamp(2, Timestamp.from(Instant.now())); + statement.setString(3, pipeline); + statement.setString(4, documentId); + statement.executeUpdate(); + } catch (SQLException e) { + throw new IllegalStateException("Tracker " + column + " update failed for '" + documentId + "'", e); + } + } + + @Override + public void markFailed(String pipeline, String documentId, String fingerprint) { + try (Connection connection = dataSource.getConnection()) { + String update = "UPDATE " + TABLE + " SET fingerprint = ?, status = 'failed', updated_at = ? " + + "WHERE pipeline = ? AND doc_id = ?"; + int updated; + try (PreparedStatement statement = connection.prepareStatement(update)) { + statement.setString(1, fingerprint); + statement.setTimestamp(2, Timestamp.from(Instant.now())); + statement.setString(3, pipeline); + statement.setString(4, documentId); + updated = statement.executeUpdate(); + } + if (updated == 0) { + String insert = "INSERT INTO " + TABLE + + " (pipeline, doc_id, fingerprint, status, origin, updated_at) " + + "VALUES (?, ?, ?, 'failed', 'source', ?)"; + try (PreparedStatement statement = connection.prepareStatement(insert)) { + statement.setString(1, pipeline); + statement.setString(2, documentId); + statement.setString(3, fingerprint); + statement.setTimestamp(4, Timestamp.from(Instant.now())); + statement.executeUpdate(); + } + } + } catch (SQLException e) { + throw new IllegalStateException("Tracker dead-letter update failed for '" + documentId + "'", e); + } + } + + @Override + public void deleteRow(String pipeline, String documentId) { + String sql = "DELETE FROM " + TABLE + " WHERE pipeline = ? AND doc_id = ?"; + try (Connection connection = dataSource.getConnection(); + PreparedStatement statement = connection.prepareStatement(sql)) { + statement.setString(1, pipeline); + statement.setString(2, documentId); + statement.executeUpdate(); + } catch (SQLException e) { + throw new IllegalStateException("Tracker row deletion failed for '" + documentId + "'", e); + } + } + + @Override + public void commit(String pipeline, String documentId, String fingerprint, String contentHash, + int segmentCount) { + String sql = "UPDATE " + TABLE + " SET fingerprint = ?, content_hash = ?, segment_count = ?, " + + "intended_count = ?, status = 'done', updated_at = ? WHERE pipeline = ? AND doc_id = ?"; Review Comment: `executeUpdate()`'s row count is never checked, so `commit` on a key with no preceding intent writes nothing and reports success. Given the two-phase protocol is the whole point of the design, a commit that quietly evaporates is a rough thing to debug — worth failing loudly. Same shape at line 274 — `refreshFingerprint` silently no-ops when the row isn't `done`. That one may well be intentional, but it's undocumented. ########## integration-tests/langchain4j-ingestion-tracker/pom.xml: ########## @@ -0,0 +1,128 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You 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. + +--> +<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 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <modelVersion>4.0.0</modelVersion> + <parent> + <groupId>org.apache.camel.quarkus</groupId> + <artifactId>camel-quarkus-build-parent-it</artifactId> + <version>3.39.0-SNAPSHOT</version> + <relativePath>../../poms/build-parent-it/pom.xml</relativePath> + </parent> + + <artifactId>camel-quarkus-integration-test-langchain4j-ingestion-tracker</artifactId> + <name>Camel Quarkus :: Integration Tests :: LangChain4j Sync Tracker</name> Review Comment: Leftover from the `sync-ledger` naming — should be "Ingestion Tracker". ########## integration-tests/langchain4j-ingestion-tracker/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingestiontracker/it/IngestionTrackerResource.java: ########## @@ -0,0 +1,135 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.camel.quarkus.component.langchain4j.ingestiontracker.it; + +import java.util.List; + +import javax.sql.DataSource; + +import jakarta.annotation.PostConstruct; +import jakarta.inject.Inject; +import jakarta.ws.rs.DELETE; +import jakarta.ws.rs.GET; +import jakarta.ws.rs.NotFoundException; +import jakarta.ws.rs.POST; +import jakarta.ws.rs.Path; +import jakarta.ws.rs.PathParam; +import jakarta.ws.rs.QueryParam; +import org.apache.camel.quarkus.component.support.langchain4j.tracker.IngestionTracker; +import org.apache.camel.quarkus.component.support.langchain4j.tracker.IngestionTracker.TrackerRow; +import org.apache.camel.quarkus.component.support.langchain4j.tracker.jdbc.JdbcIngestionTracker; + +/** + * Exercises {@link JdbcIngestionTracker} against a real datasource. A REST resource rather than a + * directly injected test field because {@code @QuarkusIntegrationTest} runs the application as a + * separate process and cannot use {@code @Inject}. + */ +@Path("/ingestion-tracker") +public class IngestionTrackerResource { + + @Inject + DataSource dataSource; + + private IngestionTracker tracker; + + @PostConstruct Review Comment: DDL in `@PostConstruct` on an unscoped JAX-RS resource. `@ApplicationScoped` with a `StartupEvent` observer would be more predictable. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
