JiriOndrusek commented on code in PR #9006: URL: https://github.com/apache/camel-quarkus/pull/9006#discussion_r3783569813
########## 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: Done — added intent→intent, commit-without-intent, markFailed-from-in_progress and refreshFingerprint-on-non-done tests. ########## 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: Done — unique pipeline id per test. ########## 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: Done — `@ApplicationScoped` with a `StartupEvent` observer. ########## 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: Done — the claim is now implemented, and an operator note on `doc_id` sensitivity was added. ########## 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: Done — renamed to "Ingestion Tracker". ########## 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: Done — renamed to `camel_quarkus_ingestion_tracker`. -- 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]
