Changeset import/export over gRPC

common/src/main/java/dev/ikm/tinkar/common/service/RemoteChangesetService.java

@@ -0,0 +1,81 @@
+/*
+ * Copyright © 2015 Integrated Knowledge Management (support@ikm.dev)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package dev.ikm.tinkar.common.service;
+
+import dev.ikm.tinkar.common.id.PublicId;
+
+import java.io.File;
+import java.util.List;
+
+/**
+ * Imports and exports changesets on a remote datastore, for a client whose own store is only a
+ * local view of it — Komet connected over gRPC.
+ *
+ * <p>Such a client cannot import into its local store: the import would be lost when it closes,
+ * and nobody else would see it. Nor can it export from it: the local store holds only what the
+ * client happened to load. Both have to happen where the data lives.
+ *
+ * <p>Found the way {@link RemoteReasonerService} is, through
+ * {@code ServiceLifecycleManager.get().getRunningService(RemoteChangesetService.class)}, so the
+ * client does not depend on the implementation.
+ */
+public interface RemoteChangesetService {
+
+    /** Notified as a remote job advances. */
+    @FunctionalInterface
+    interface ProgressListener {
+        /**
+         * @param done    work done so far; negative while the total is not known
+         * @param total   total work; 0 or 1 while it is not known
+         * @param message what the job is doing, or why it is waiting
+         */
+        void onProgress(long done, long total, String message);
+    }
+
+    /**
+     * Uploads {@code changeset} and imports it on the remote datastore, returning once the import
+     * has finished.
+     *
+     * <p>A remote import cannot be cancelled — stopped part way it would leave some entities
+     * written. Cancelling {@code tracker} only stops waiting for it; the import carries on.
+     *
+     * @param tracker cancelling it stops waiting; may be null
+     * @return the counts of what was imported
+     * @throws java.util.concurrent.CancellationException if {@code tracker} was cancelled first
+     * @throws IllegalStateException if the remote import failed
+     */
+    EntityCountSummary importChangeset(File changeset, ProgressListener listener, TrackingCallable<?> tracker);
+
+    /**
+     * Exports the entities changed between two times — a "change set" — from the remote datastore
+     * into {@code target}.
+     *
+     * @param tracker cancelling it cancels the remote export, which leaves no file; may be null
+     * @return the counts of what was exported
+     * @throws java.util.concurrent.CancellationException if the export was cancelled
+     * @throws IllegalStateException if the remote export failed
+     */
+    EntityCountSummary exportChangeSet(File target, long fromEpochMillis, long toEpochMillis,
+                                       ProgressListener listener, TrackingCallable<?> tracker);
+
+    /**
+     * Exports the members of {@code membershipTags} from the remote datastore into {@code target}.
+     *
+     * @see #exportChangeSet
+     */
+    EntityCountSummary exportMembership(File target, List<PublicId> membershipTags,
+                                        ProgressListener listener, TrackingCallable<?> tracker);
+}

common/src/main/java/dev/ikm/tinkar/common/util/thread/StructuredScopes.java

@@ -1,7 +1,9 @@
 package dev.ikm.tinkar.common.util.thread;
 
+import java.util.concurrent.Semaphore;
 import java.util.concurrent.StructuredTaskScope;
 import java.util.concurrent.StructuredTaskScope.Joiner;
+import java.util.concurrent.TimeUnit;
 
 /**
  * Opens {@link StructuredTaskScope}s with the failure behaviour IKE code was
@@ -38,4 +40,29 @@ public static <T> StructuredTaskScope<T, Void, SubtaskFailedException> open() {
         return StructuredTaskScope.open(Joiner.<T, SubtaskFailedException>awaitAllSuccessfulOrThrow(
                 SubtaskFailedException::new));
     }
+
+    /** How often a wait for a permit checks whether its scope has been cancelled. */
+    private static final long PERMIT_POLL_MS = 100L;
+
+    /**
+     * Takes a permit for the next subtask, or returns false once {@code scope} has been cancelled.
+     *
+     * <p>A scope from {@link #open()} cancels itself on the first failed subtask, and a cancelled
+     * scope does not run the subtasks forked after that. Each of those would have released its
+     * permit when it finished, so their permits never come back: a plain {@code acquire()} then
+     * blocks for good, and the work hangs instead of reporting the failure. Polling lets the
+     * forking thread notice the cancellation and stop, so that {@code join()} can throw the
+     * subtask's error.
+     *
+     * @return true with a permit taken; false, with none, once the scope is cancelled
+     */
+    public static boolean acquireUnlessCancelled(Semaphore permits, StructuredTaskScope<?, ?, ?> scope)
+            throws InterruptedException {
+        while (!permits.tryAcquire(PERMIT_POLL_MS, TimeUnit.MILLISECONDS)) {
+            if (scope.isCancelled()) {
+                return false;
+            }
+        }
+        return true;
+    }
 }

common/src/test/java/dev/ikm/tinkar/common/util/thread/StructuredScopesTest.java

@@ -2,13 +2,18 @@
 
 import org.junit.jupiter.api.Test;
 
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.Semaphore;
 import java.util.concurrent.StructuredTaskScope;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertInstanceOf;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 /**
  * The JDK 25 {@code StructuredTaskScope.open()} failure contract that
@@ -46,4 +51,37 @@ void aFailingSubtask_makesJoinThrowAnUncheckedExceptionWithTheCause() {
         assertInstanceOf(RuntimeException.class, thrown, "callers need no checked-exception handling");
         assertEquals(failure, thrown.getCause());
     }
+
+    @Test
+    void aPermitIsTakenWhileTheScopeRuns() throws Exception {
+        Semaphore permits = new Semaphore(1);
+        try (StructuredTaskScope<Object, Void, SubtaskFailedException> scope = StructuredScopes.open()) {
+            assertTrue(StructuredScopes.acquireUnlessCancelled(permits, scope));
+            scope.join();
+        }
+        assertEquals(0, permits.availablePermits(), "the permit was taken");
+    }
+
+    /**
+     * Readers fork a subtask per record and take a permit for each. When one fails, the scope
+     * cancels itself and never runs the subtasks forked after it, so their permits never come
+     * back; a plain {@code acquire()} then blocked for good, and an import or export hung
+     * instead of reporting the failure.
+     */
+    @Test
+    void aWaitForAPermit_endsOnceAFailedSubtaskCancelsTheScope() {
+        assertTimeoutPreemptively(Duration.ofSeconds(10), () -> {
+            Semaphore permits = new Semaphore(1);
+            try (StructuredTaskScope<Object, Void, SubtaskFailedException> scope = StructuredScopes.open()) {
+                // Holds the only permit and fails without releasing it, as a subtask the
+                // cancelled scope never ran would.
+                assertTrue(StructuredScopes.acquireUnlessCancelled(permits, scope));
+                scope.fork(() -> {
+                    throw new IllegalStateException("a record failed");
+                });
+                assertFalse(StructuredScopes.acquireUnlessCancelled(permits, scope));
+                assertThrows(SubtaskFailedException.class, scope::join);
+            }
+        });
+    }
 }

entity/src/main/java/dev/ikm/tinkar/entity/aggregator/DefaultEntityAggregator.java

@@ -19,11 +19,8 @@
 
 import dev.ikm.tinkar.common.service.internal.EntityStore;
 import dev.ikm.tinkar.common.service.EntityCountSummary;
-import dev.ikm.tinkar.common.service.PrimitiveData;
 import dev.ikm.tinkar.entity.ConceptEntity;
 import dev.ikm.tinkar.entity.Entity;
-import dev.ikm.tinkar.entity.EntityHandle;
-import dev.ikm.tinkar.entity.EntityService;
 import dev.ikm.tinkar.entity.PatternEntity;
 import dev.ikm.tinkar.entity.SemanticEntity;
 import dev.ikm.tinkar.entity.StampEntity;
@@ -32,11 +29,16 @@
 
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Consumer;
-import java.util.function.IntConsumer;
 
 public class DefaultEntityAggregator extends EntityAggregator {
     private static final Logger LOG = LoggerFactory.getLogger(DefaultEntityAggregator.class);
 
+    /** Cheap: counting only iterates nids, it never reads an entity. */
+    @Override
+    public boolean totalCountIsCheap() {
+        return true;
+    }
+
     @Override
     public EntityCountSummary aggregate(LongConsumer nidConsumer) {
         initCounts();
@@ -83,9 +85,13 @@ public EntityCountSummary aggregateEntities(Consumer<Entity<?>> entityConsumer)
         AtomicLong orphanCount = new AtomicLong();
 
         EntityStore.current().forEachStampNid(nid -> dispatch(nid, entityConsumer, orphanCount));
+        throwIfCancelled();
         EntityStore.current().forEachConceptNid(nid -> dispatch(nid, entityConsumer, orphanCount));
+        throwIfCancelled();
         EntityStore.current().forEachSemanticNid(nid -> dispatch(nid, entityConsumer, orphanCount));
+        throwIfCancelled();
         EntityStore.current().forEachPatternNid(nid -> dispatch(nid, entityConsumer, orphanCount));
+        throwIfCancelled();
 
         long orphans = orphanCount.get();
         if (orphans > 0) {
@@ -97,7 +103,12 @@ public EntityCountSummary aggregateEntities(Consumer<Entity<?>> entityConsumer)
     private void dispatch(long nid,
                           Consumer<Entity<?>> entityConsumer,
                           AtomicLong orphanCount) {
-        Entity<?> entity = EntityHandle.get(nid).orNull();
+        if (isCancelled()) {
+            return;
+        }
+        // Past the entity cache: a full export reads every entity once, and through the cache it
+        // would evict the working set and stall on the eviction lock.
+        Entity<?> entity = readUncached(nid);
         if (entity == null) {
             orphanCount.incrementAndGet();
             return;

entity/src/main/java/dev/ikm/tinkar/entity/aggregator/EntityAggregator.java

@@ -18,15 +18,16 @@
 import java.util.function.LongConsumer;
 
 import dev.ikm.tinkar.common.service.EntityCountSummary;
+import dev.ikm.tinkar.common.service.internal.EntityStore;
 import dev.ikm.tinkar.entity.Entity;
-import dev.ikm.tinkar.entity.EntityHandle;
-import dev.ikm.tinkar.entity.EntityService;
+import dev.ikm.tinkar.entity.EntityRecordFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.concurrent.CancellationException;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BooleanSupplier;
 import java.util.function.Consumer;
-import java.util.function.IntConsumer;
 
 public abstract class EntityAggregator {
     private static final Logger LOG = LoggerFactory.getLogger(EntityAggregator.class);
@@ -36,16 +37,62 @@ public abstract class EntityAggregator {
     protected AtomicLong patternsAggregatedCount = new AtomicLong(0);
     protected AtomicLong stampsAggregatedCount = new AtomicLong(0);
 
+    /** Checked during a scan; once true, the scan skips what is left and the aggregation throws. */
+    private volatile BooleanSupplier cancelled = () -> false;
+
     public abstract EntityCountSummary aggregate(LongConsumer nidConsumer);
 
     /**
-     * Aggregates entities by resolving each nid produced by {@link #aggregate(IntConsumer)}
+     * Makes the aggregation stop early once {@code cancelled} returns true, throwing
+     * {@link CancellationException}. Scans run in parallel and cannot be broken out of, so each
+     * remaining nid is skipped instead — it is reading them that takes the time.
+     */
+    public void setCancellationCheck(BooleanSupplier cancelled) {
+        this.cancelled = cancelled == null ? () -> false : cancelled;
+    }
+
+    protected boolean isCancelled() {
+        return cancelled.getAsBoolean();
+    }
+
+    /** @throws CancellationException if the aggregation has been cancelled */
+    protected void throwIfCancelled() {
+        if (isCancelled()) {
+            throw new CancellationException("Aggregation cancelled");
+        }
+    }
+
+    /**
+     * Whether {@link #totalCount()} is cheap enough to call before aggregating, to size a progress
+     * bar. It runs the whole aggregation once over, so for an aggregator that has to read every
+     * entity to decide — a time range, say — it doubles the cost of an export.
+     */
+    public boolean totalCountIsCheap() {
+        return false;
+    }
+
+    /**
+     * Reads an entity straight from the store, bypassing the shared entity cache.
+     *
+     * <p>For scans that touch every entity once: sending millions of one-off reads through the
+     * cache evicts the working set everyone else relies on, and with many threads it stalls on
+     * the cache's eviction lock.
+     *
+     * @return the entity, or null if the nid has no entity bytes
+     */
+    protected static Entity<?> readUncached(long nid) {
+        byte[] bytes = EntityStore.current().getBytes(nid);
+        return bytes == null ? null : EntityRecordFactory.make(bytes);
+    }
+
+    /**
+     * Aggregates entities by resolving each nid produced by {@link #aggregate(LongConsumer)}
      * to its {@link Entity} and forwarding non-null results to {@code entityConsumer}.
      * Orphan nids — sequences allocated by the store but with no committed entity bytes
      * (typically a component that was canceled between sequence allocation and commit) —
      * are silently skipped and reported in aggregate at INFO level.
      *
-     * <p>The default implementation delegates to {@link #aggregate(IntConsumer)} and
+     * <p>The default implementation delegates to {@link #aggregate(LongConsumer)} and
      * therefore inherits whatever per-bucket counting that method performs. With the
      * {@link DefaultEntityAggregator} that means per-bucket counts will overstate by
      * the number of orphans, since they are incremented per nid visited rather than
@@ -59,13 +106,17 @@ public abstract class EntityAggregator {
     public EntityCountSummary aggregateEntities(Consumer<Entity<?>> entityConsumer) {
         AtomicLong orphanCount = new AtomicLong();
         EntityCountSummary summary = aggregate((long nid) -> {
-            Entity<?> entity = EntityHandle.get(nid).orNull();
+            if (isCancelled()) {
+                return;
+            }
+            Entity<?> entity = readUncached(nid);
             if (entity == null) {
                 orphanCount.incrementAndGet();
                 return;
             }
             entityConsumer.accept(entity);
         });
+        throwIfCancelled();
         long orphans = orphanCount.get();
         if (orphans > 0) {
             LOG.info("Skipped {} orphan nid(s) during aggregation (allocated, no entity bytes)", orphans);