Concurrent UPDATE of multi-page records (large embeddings) with LSM_VECTOR index causes "Invalid pointer to a chunk" and "Concurrent modification on page" errors.

[syntact-io-office-user]

Our ingestion pipeline corrupts our database consistently when concurrency is high enough.

the corrupted records cannot be retrieved and also not deleted. Also sometime the previous or next records throw class cast exceptions (probably because the record borders got messed up)

edit
sorry I was too quick with the test case. this doesn't reproduce the corruption consistently. I will get back when I have a proper testcase

[lvca]

Ok, let me know, checking your other issue #3115

[syntact-io-office-user]

it is a bit clearer now, but still not 100%. What we know so far is that under heavy concurrency there are 2 types of errors arising.

1. All sort of missing pointer to chunk / record was deleted errors (even though our code does not delete records)
2. Error on reading multi page record

we cannot reproduce problem number 2 in a self contained testfile, but problem number one can be reproduced with the following test :

package com.acme;

import com.arcadedb.ContextConfiguration;
import com.arcadedb.GlobalConfiguration;
import com.arcadedb.database.Database;
import com.arcadedb.engine.ComponentFile;
import com.arcadedb.exception.DatabaseOperationException;
import com.arcadedb.query.sql.executor.ResultSet;
import com.arcadedb.schema.Schema;
import com.arcadedb.schema.Type;
import com.arcadedb.schema.VertexType;
import com.arcadedb.server.ArcadeDBServer;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

import java.io.File;
import java.util.Random;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Minimal test to reproduce multi-page record corruption in ArcadeDB.
 *
 * Reproduces: "Invalid pointer to a chunk for record #X:Y"
 *
 * Scenario:
 * 1. Create records with large embeddings (multi-page records)
 * 2. Concurrently update embeddings from many threads
 * 3. Corruption occurs: invalid pointer errors when reading/writing
 */
public class MinimalMultiPageCorruptionTest {

    private ArcadeDBServer server;
    private static final String DB_DIR = "./databases-test/minimal-corruption";
    private static final String DB_NAME = "test_db";

    // Minimal parameters to reproduce
    private static final int NUM_RECORDS = 10000;  // More records to increase contention
    private static final int CONCURRENT_THREADS = 500;
    private static final int EMBEDDING_DIM = 3072;  // Multi-page record size

    private final Random random = new Random();
    private final AtomicInteger errors = new AtomicInteger(0);

    @BeforeEach
    void setUp() {
        deleteDirectory(new File(DB_DIR));

        ContextConfiguration config = new ContextConfiguration();
        config.setValue(GlobalConfiguration.SERVER_DATABASE_DIRECTORY, DB_DIR);
        config.setValue(GlobalConfiguration.SERVER_ROOT_PATH, ".");
        config.setValue(GlobalConfiguration.HA_ENABLED, false);

        server = new ArcadeDBServer(config);
        server.start();
        server.createDatabase(DB_NAME, ComponentFile.MODE.READ_WRITE);

        Database db = server.getDatabase(DB_NAME);
        db.transaction(() -> {
            Schema schema = db.getSchema();

            // Parent type with embedding (like ContentV in production)
            VertexType baseV = schema.createVertexType("BaseV");
            baseV.createProperty("id", Type.STRING);
            baseV.createProperty("embedding", Type.ARRAY_OF_FLOATS);

            // Child type (like TextV in production)
            VertexType recordV = schema.createVertexType("RecordV");
            recordV.addSuperType(baseV);  // Inheritance is key!

            // Unique index on id
            baseV.createTypeIndex(Schema.INDEX_TYPE.LSM_TREE, true, "id");

            // LSM Vector Index on PARENT type (like production)
            db.command("sql", """
                CREATE INDEX IF NOT EXISTS ON BaseV (embedding) LSM_VECTOR
                METADATA {
                    "dimensions": %d,
                    "similarity": "COSINE"
                }""".formatted(EMBEDDING_DIM));
        });
    }

    @AfterEach
    void tearDown() {
        if (server != null) {
            server.stop();
        }
        deleteDirectory(new File(DB_DIR));
    }

    @Test
    void testMultiPageRecordCorruption() throws Exception {
        Database db = server.getDatabase(DB_NAME);

        // Phase 1: Create records with zero embeddings
        System.out.println("Creating " + NUM_RECORDS + " records...");
        for (int i = 0; i < NUM_RECORDS; i++) {
            String id = "record_" + i;
            float[] zeroEmbedding = new float[EMBEDDING_DIM];
            db.transaction(() -> {
                db.command("sql", "INSERT INTO RecordV SET id=?, embedding=?", id, zeroEmbedding);
            }, false, 3);
        }

        // Phase 2: Concurrently update embeddings
        System.out.println("Updating embeddings with " + CONCURRENT_THREADS + " concurrent threads...");
        ExecutorService executor = Executors.newFixedThreadPool(CONCURRENT_THREADS);
        CountDownLatch latch = new CountDownLatch(NUM_RECORDS);

        for (int i = 0; i < NUM_RECORDS; i++) {
            String id = "record_" + i;
            executor.submit(() -> {
                try {
                    updateEmbedding(db, id);
                } catch (Exception e) {
                    // Expected - some may fail
                } finally {
                    latch.countDown();
                }
            });
        }

        latch.await(5, TimeUnit.MINUTES);
        executor.shutdown();
        executor.awaitTermination(1, TimeUnit.MINUTES);

        // Phase 3: Verification - try to read all embeddings to detect corruption
        System.out.println("Verifying all records...");
        for (int i = 0; i < NUM_RECORDS; i++) {
            String id = "record_" + i;
            try {
                db.transaction(() -> {
                    try (ResultSet rs = db.query("sql", "SELECT embedding FROM RecordV WHERE id=?", id)) {
                        if (rs.hasNext()) {
                            rs.next().getProperty("embedding"); // Force read
                        }
                    }
                }, true, 1);
            } catch (DatabaseOperationException e) {
                if (e.getMessage().contains("Invalid pointer") || e.getMessage().contains("was deleted")) {
                    errors.incrementAndGet();
                    System.err.println("✗ Corruption detected on " + id + ": " + e.getMessage());
                }
            }
        }

        // Results
        System.out.println("\n=== RESULTS ===");
        System.out.println("Total records: " + NUM_RECORDS);
        System.out.println("Corruption errors: " + errors.get());

        if (errors.get() > 0) {
            System.out.println("\n✓ SUCCESS: Reproduced corruption with " + errors.get() + " errors!");
        } else {
            System.out.println("\n⚠ No corruption detected. Try running again.");
        }
    }

    private void updateEmbedding(Database db, String id) {
        try {
            db.transaction(() -> {
                // Read current record
                try (ResultSet rs = db.query("sql", "SELECT embedding FROM RecordV WHERE id=?", id)) {
                    if (!rs.hasNext()) return;
                    rs.next().getProperty("embedding");
                }

                // Update with new embedding (multi-page write)
                float[] newEmbedding = randomEmbedding();
                db.command("sql", "UPDATE RecordV SET embedding=? WHERE id=?", newEmbedding, id);
            }, false, 3);
        } catch (DatabaseOperationException e) {
            if (e.getMessage().contains("Invalid pointer") || e.getMessage().contains("was deleted")) {
                errors.incrementAndGet();
                System.err.println("✗ Corruption error on " + id + ": " + e.getMessage());
            }
        }
    }

    private float[] randomEmbedding() {
        float[] embedding = new float[EMBEDDING_DIM];
        for (int i = 0; i < EMBEDDING_DIM; i++) {
            embedding[i] = random.nextFloat();
        }
        return embedding;
    }

    private void deleteDirectory(File dir) {
        if (dir.exists()) {
            File[] files = dir.listFiles();
            if (files != null) {
                for (File file : files) {
                    if (file.isDirectory()) {
                        deleteDirectory(file);
                    } else {
                        file.delete();
                    }
                }
            }
            dir.delete();
        }
    }
}

[lvca]

Can you please try with REPEATABLE_READ transaction isolation? Set it at database level: db.setTransactionIsolationLevel(REPEATABLE_READ).

[lvca]

I've also update a part of code that should be helpful for this use case, let me know if you can also retest against the latest main branch.

[syntact-io-office-user]

thank you so much for your support. We havent so far used arcade directly from source (we will try that soon)

Changing the isolation level helped with the test case. Unfortunately in production it didn't really help. also it resulted in lots of those

com.arcadedb.exception.TransactionException: Transaction error on commit
at com.arcadedb.database.TransactionContext.commit1stPhase(TransactionContext.java:653) ~[arcadedb-engine-25.12.1.jar!/:na]
at com.arcadedb.database.TransactionContext.commit(TransactionContext.java:128) ~[arcadedb-engine-25.12.1.jar!/:na]
at com.arcadedb.database.LocalDatabase.lambda$commit$5(LocalDatabase.java:431) ~[arcadedb-engine-25.12.1.jar!/:na]
at com.arcadedb.database.LocalDatabase.executeInReadLock(LocalDatabase.java:1467) ~[arcadedb-engine-25.12.1.jar!/:na]
at com.arcadedb.database.LocalDatabase.commit(LocalDatabase.java:426) ~[arcadedb-engine-25.12.1.jar!/:na]
at com.arcadedb.database.LocalDatabase.transaction(LocalDatabase.java:1110) ~[arcadedb-engine-25.12.1.jar!/:na]
at com.arcadedb.database.LocalDatabase.transaction(LocalDatabase.java:1082) ~[arcadedb-engine-25.12.1.jar!/:na]
at com.arcadedb.server.ServerDatabase.transaction(ServerDatabase.java:359) ~[arcadedb-server-25.12.1.jar!/:na]
at io.syntact.index.repository.GraphRepository.storeEntitiesAndRelationshipsForTextChunk(GraphRepository.java:747) ~[!/:0.0.1-SNAPSHOT]
at io.syntact.index.service.processing.processors.TextProcessor.process(TextProcessor.java:73) ~[!/:0.0.1-SNAPSHOT]
at io.syntact.index.service.processing.ContentProcessingDispatcher.dispatchToProcessor(ContentProcessingDispatcher.java:160) ~[!/:0.0.1-SNAPSHOT]
at io.syntact.index.service.processing.ContentProcessingDispatcher.lambda$processContent$0(ContentProcessingDispatcher.java:70) ~[!/:0.0.1-SNAPSHOT]
at io.syntact.index.service.processing.ContentProcessingDispatcher.processWithRetry(ContentProcessingDispatcher.java:177) ~[!/:0.0.1-SNAPSHOT]
at io.syntact.index.service.processing.ContentProcessingDispatcher.processContent(ContentProcessingDispatcher.java:70) ~[!/:0.0.1-SNAPSHOT]
at io.syntact.index.service.processing.ContentProcessingDispatcher.lambda$processContentBatch$0(ContentProcessingDispatcher.java:104) ~[!/:0.0.1-SNAPSHOT]
at java.base/java.util.concurrent.CompletableFuture$AsyncRun.run(Unknown Source) ~[na:na]
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source) ~[na:na]
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source) ~[na:na]
at java.base/java.lang.Thread.run(Unknown Source) ~[na:na]
Caused by: com.arcadedb.exception.TimeoutException: Timeout on locking file 1 (EntityV_0.1.65536.v0.bucket) during commit (fileIds=[1, 2, 3, 16, 17, 22, 23, 34, 103, 105, 122, 124, 126], timeout=5000ms
at com.arcadedb.engine.TransactionManager.tryLockFiles(TransactionManager.java:496) ~[arcadedb-engine-25.12.1.jar!/:na]
at com.arcadedb.database.TransactionContext.lockFilesInOrder(TransactionContext.java:828) ~[arcadedb-engine-25.12.1.jar!/:na]
at com.arcadedb.database.TransactionContext.commit1stPhase(TransactionContext.java:587) ~[arcadedb-engine-25.12.1.jar!/:na]
... 18 common frames omitted

finaly the application crashed.
What was also weird was that the ingestion status kept flashing back to an old state showing 17%, 18% then again 7% for a long time flashing back to 19% over 5 Minutes. No transaction is open that long.

i was thinking if i should extract the embedding in a dedicated node. we have many status update during ingestions on our nodes and i feel like that in combination with the large embeddings (even written once) have something to do with this.