diff --git a/common/src/main/java/dev/ikm/tinkar/common/service/RemoteChangesetService.java b/common/src/main/java/dev/ikm/tinkar/common/service/RemoteChangesetService.java new file mode 100644 index 000000000..c7fb932a2 --- /dev/null +++ b/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. + * + *

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. + * + *

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. + * + *

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 membershipTags, + ProgressListener listener, TrackingCallable tracker); +} diff --git a/common/src/main/java/dev/ikm/tinkar/common/util/thread/StructuredScopes.java b/common/src/main/java/dev/ikm/tinkar/common/util/thread/StructuredScopes.java index 908e07368..9e5245495 100644 --- a/common/src/main/java/dev/ikm/tinkar/common/util/thread/StructuredScopes.java +++ b/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 StructuredTaskScope open() { return StructuredTaskScope.open(Joiner.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. + * + *

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; + } } diff --git a/common/src/test/java/dev/ikm/tinkar/common/util/thread/StructuredScopesTest.java b/common/src/test/java/dev/ikm/tinkar/common/util/thread/StructuredScopesTest.java index b2827ff5c..f0e613b86 100644 --- a/common/src/test/java/dev/ikm/tinkar/common/util/thread/StructuredScopesTest.java +++ b/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 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 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); + } + }); + } } diff --git a/entity/src/main/java/dev/ikm/tinkar/entity/aggregator/DefaultEntityAggregator.java b/entity/src/main/java/dev/ikm/tinkar/entity/aggregator/DefaultEntityAggregator.java index 95c5d53a7..6ed843526 100644 --- a/entity/src/main/java/dev/ikm/tinkar/entity/aggregator/DefaultEntityAggregator.java +++ b/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> 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> entityConsumer) private void dispatch(long nid, Consumer> 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; diff --git a/entity/src/main/java/dev/ikm/tinkar/entity/aggregator/EntityAggregator.java b/entity/src/main/java/dev/ikm/tinkar/entity/aggregator/EntityAggregator.java index d0257a51b..efe257e6d 100644 --- a/entity/src/main/java/dev/ikm/tinkar/entity/aggregator/EntityAggregator.java +++ b/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. + * + *

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. * - *

The default implementation delegates to {@link #aggregate(IntConsumer)} and + *

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> 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); diff --git a/entity/src/main/java/dev/ikm/tinkar/entity/aggregator/TemporalEntityAggregator.java b/entity/src/main/java/dev/ikm/tinkar/entity/aggregator/TemporalEntityAggregator.java index 0af321537..fe5c24462 100644 --- a/entity/src/main/java/dev/ikm/tinkar/entity/aggregator/TemporalEntityAggregator.java +++ b/entity/src/main/java/dev/ikm/tinkar/entity/aggregator/TemporalEntityAggregator.java @@ -18,20 +18,18 @@ import java.util.function.LongConsumer; import dev.ikm.tinkar.common.service.internal.EntityStore; -import dev.ikm.tinkar.common.service.PrimitiveData; import dev.ikm.tinkar.common.service.EntityCountSummary; import dev.ikm.tinkar.entity.Entity; -import dev.ikm.tinkar.entity.EntityHandle; import dev.ikm.tinkar.entity.EntityService; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.ArrayList; import java.util.Collections; -import java.util.HashSet; import java.util.List; import java.util.Set; -import java.util.function.IntConsumer; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicLong; public class TemporalEntityAggregator extends EntityAggregator { private static final Logger LOG = LoggerFactory.getLogger(TemporalEntityAggregator.class); @@ -39,12 +37,12 @@ public class TemporalEntityAggregator extends EntityAggregator { private final long fromEpochMillis; private final long toEpochMillis; - /** Orphan nids seen by the last {@link #aggregate(IntConsumer)} run — sequences + /** Orphan nids seen by the last {@link #aggregate(LongConsumer)} run — sequences * allocated by the store but with no committed entity bytes (typically canceled * between allocation and commit). Counted for the log line; never counted in the * summary and never emitted, so a manifest written from the summary always matches * the records actually delivered (IKE-Network/ike-issues#933). */ - private long lastOrphanCount; + private final AtomicLong lastOrphanCount = new AtomicLong(); public TemporalEntityAggregator(long fromEpochMillis, long toEpochMillis) { this.fromEpochMillis = fromEpochMillis; @@ -54,94 +52,86 @@ public TemporalEntityAggregator(long fromEpochMillis, long toEpochMillis) { @Override public EntityCountSummary aggregate(LongConsumer nidConsumer) { initCounts(); + // Concurrent collections throughout: the forEach*Nid scans run in parallel, and a plain + // HashSet or ArrayList filled from several threads can drop entries — here, stamps the + // export would then silently leave out. // Filter Stamp Nids based on the supplied time span - Set filteredStampNids = new HashSet<>(); + Set filteredStampNids = ConcurrentHashMap.newKeySet(); EntityStore.current().forEachStampNid((stampNid) -> { + if (isCancelled()) { + return; + } EntityService.get().getStamp(stampNid).ifPresent((stampEntity) -> { if (fromEpochMillis <= stampEntity.time() && stampEntity.time() <= toEpochMillis) { filteredStampNids.add(stampEntity.nid()); } }); }); + throwIfCancelled(); - List stampsToExport = new ArrayList<>(); - - // Aggregate concepts with a filtered stamp. Resolution goes through - // EntityHandle — the byte-backed lookup the downstream consumer uses — so a - // nid is counted if and only if it can actually be delivered: an orphan nid - // (allocated, no committed bytes; the entity cache may still answer for it) - // is excluded from the count, the emission, and the stamp collection alike - // (IKE-Network/ike-issues#933). - lastOrphanCount = 0; - EntityStore.current().forEachConceptNid((conceptNid) -> { - Entity conceptEntity = EntityHandle.get(conceptNid).orNull(); - if (conceptEntity == null) { - lastOrphanCount++; - return; - } - Set conceptStampNidList = conceptEntity.stampNids().mapToSet(i->i); - // Write whole chronology if ANY of the stamps satisfy conditions - if (!Collections.disjoint(filteredStampNids, conceptStampNidList)) { - conceptsAggregatedCount.incrementAndGet(); - nidConsumer.accept(conceptNid); - stampsToExport.addAll(conceptStampNidList); - } - }); + Set stampsToExport = ConcurrentHashMap.newKeySet(); - // Aggregate semantics with a filtered stamp - EntityStore.current().forEachSemanticNid((semanticNid) -> { - Entity semanticEntity = EntityHandle.get(semanticNid).orNull(); - if (semanticEntity == null) { - lastOrphanCount++; - return; - } - Set semanticStampNidList = semanticEntity.stampNids().mapToSet(i->i); - // Write whole chronology if ANY of the stamps satisfy conditions - if (!Collections.disjoint(filteredStampNids, semanticStampNidList)) { - semanticsAggregatedCount.incrementAndGet(); - nidConsumer.accept(semanticNid); - stampsToExport.addAll(semanticStampNidList); - } - }); + // Every concept, semantic and pattern is read to see whether any of its stamps falls in + // the range — there is no index from a stamp to the components that use it. Read past the + // entity cache (readUncached): a scan touching every entity once would otherwise evict + // the cache's working set and, from many threads, stall on its eviction lock. + // + // A nid is counted if and only if it can actually be delivered: an orphan nid (allocated, + // no committed bytes) is excluded from the count, the emission, and the stamp collection + // alike (IKE-Network/ike-issues#933). + lastOrphanCount.set(0); + EntityStore.current().forEachConceptNid(nid -> aggregateIfInRange(nid, filteredStampNids, stampsToExport, + conceptsAggregatedCount, nidConsumer)); + throwIfCancelled(); - // Aggregate patterns with a filtered stamp - EntityStore.current().forEachPatternNid((patternNid) -> { - Entity patternEntity = EntityHandle.get(patternNid).orNull(); - if (patternEntity == null) { - lastOrphanCount++; - return; - } - Set patternStampNidList = patternEntity.stampNids().mapToSet(i->i); - // Write whole chronology if ANY of the stamps satisfy conditions - if (!Collections.disjoint(filteredStampNids, patternStampNidList)) { - patternsAggregatedCount.incrementAndGet(); - nidConsumer.accept(patternNid); - stampsToExport.addAll(patternStampNidList); - } - }); + EntityStore.current().forEachSemanticNid(nid -> aggregateIfInRange(nid, filteredStampNids, stampsToExport, + semanticsAggregatedCount, nidConsumer)); + throwIfCancelled(); + + EntityStore.current().forEachPatternNid(nid -> aggregateIfInRange(nid, filteredStampNids, stampsToExport, + patternsAggregatedCount, nidConsumer)); + throwIfCancelled(); // A stamp in the window is exported whether or not a version uses it: a set may // declare one that no version does, such as the non-existent stamp. stampsToExport.addAll(filteredStampNids); - // Deduplicate and export aggregated stamps — resolution-checked like every - // other bucket, so the count only claims stamps that can be delivered. - Set deduplicatedStampsToExport = new HashSet<>(stampsToExport); + // Export the aggregated stamps — resolution-checked like every other bucket, so the count + // only claims stamps that can be delivered. List deliverableStampNids = new ArrayList<>(); - for (long stampNid : deduplicatedStampsToExport) { - if (EntityHandle.get(stampNid).isPresent()) { + for (long stampNid : stampsToExport) { + if (readUncached(stampNid) != null) { deliverableStampNids.add(stampNid); } else { - lastOrphanCount++; + lastOrphanCount.incrementAndGet(); } } stampsAggregatedCount.set(deliverableStampNids.size()); deliverableStampNids.forEach(nidConsumer::accept); - if (lastOrphanCount > 0) { + if (lastOrphanCount.get() > 0) { LOG.warn("Temporal aggregation skipped {} orphan nid(s) (allocated, no entity" - + " bytes) — excluded from counts and emission alike", lastOrphanCount); + + " bytes) — excluded from counts and emission alike", lastOrphanCount.get()); } return summarize(); } + + /** Emits {@code nid} — its whole chronology — if any of its stamps falls in the range. */ + private void aggregateIfInRange(long nid, Set filteredStampNids, Set stampsToExport, + AtomicLong aggregatedCount, LongConsumer nidConsumer) { + if (isCancelled()) { + return; + } + Entity entity = readUncached(nid); + if (entity == null) { + lastOrphanCount.incrementAndGet(); + return; + } + Set stampNids = entity.stampNids().mapToSet(i -> i); + if (!Collections.disjoint(filteredStampNids, stampNids)) { + aggregatedCount.incrementAndGet(); + nidConsumer.accept(nid); + stampsToExport.addAll(stampNids); + } + } } diff --git a/entity/src/main/java/dev/ikm/tinkar/entity/export/ExportEntitiesToProtobufFile.java b/entity/src/main/java/dev/ikm/tinkar/entity/export/ExportEntitiesToProtobufFile.java index 14650ba7a..59b3d20d1 100644 --- a/entity/src/main/java/dev/ikm/tinkar/entity/export/ExportEntitiesToProtobufFile.java +++ b/entity/src/main/java/dev/ikm/tinkar/entity/export/ExportEntitiesToProtobufFile.java @@ -58,6 +58,7 @@ import java.security.MessageDigest; import java.util.ArrayList; import java.util.HexFormat; +import java.util.concurrent.CancellationException; import java.util.concurrent.Semaphore; import java.util.concurrent.StructuredTaskScope; import java.util.zip.CRC32; @@ -94,6 +95,7 @@ import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; +import java.util.function.BooleanSupplier; import java.util.function.Consumer; import java.util.stream.Collectors; import java.util.zip.ZipEntry; @@ -149,6 +151,38 @@ public ExportEntitiesToProtobufFile(File file, List membershipTags) { updateTitle("Tag-Based Export to Protobuf"); } + /** Cancels the export when true, alongside {@link #cancel()}; for a caller that owns its own handle. */ + private volatile BooleanSupplier externalCancel = () -> false; + + /** + * Also stops the export once {@code cancelled} returns true — for a caller tracking + * cancellation with a handle of its own, such as a server job, rather than with this task's + * {@link #cancel()}. + */ + public void cancelWhen(BooleanSupplier cancelled) { + this.externalCancel = cancelled == null ? () -> false : cancelled; + } + + private boolean exportCancelled() { + return isCancelled() || externalCancel.getAsBoolean(); + } + + private static boolean causedByCancellation(Throwable e) { + for (Throwable cause = e; cause != null; cause = cause.getCause()) { + if (cause instanceof CancellationException) { + return true; + } + } + return false; + } + + /** @throws CancellationException if the export has been cancelled */ + private void throwIfCancelled() { + if (exportCancelled()) { + throw new CancellationException("Export cancelled"); + } + } + /** * Writes a format-3 changeset (IKE-Network/ike-issues#1275): the component table first, * then one gzip entry of records per pattern, in table order, written in parallel; then the @@ -159,10 +193,13 @@ public ExportEntitiesToProtobufFile(File file, List membershipTags) { public EntityCountSummary compute() { updateMessage("Analyzing Entities..."); updateProgress(-1, 1); + // Checked by the aggregator while it enumerates, then here between phases and per record. + entityAggregator.setCancellationCheck(this::exportCancelled); EntityCountSummary entityCountSummary = null; try { long started = System.nanoTime(); Buckets buckets = Buckets.collect(entityAggregator); + throwIfCancelled(); EntityCountSummary aggregated = buckets.summary(); addToTotalWork(aggregated.getTotalCount() * 2L); SequenceAllocator sequences = new SequenceAllocator(buckets); @@ -172,9 +209,11 @@ public EntityCountSummary compute() { ZipOutputStream zos = new ZipOutputStream(bos)) { updateMessage("Writing the component table..."); writeComponentTable(zos, buckets, sequences); + throwIfCancelled(); long tabled = System.nanoTime(); updateMessage("Exporting Entities..."); List written = writeRecordEntries(zos, buckets, sequences); + throwIfCancelled(); long recorded = System.nanoTime(); writeReferenceTable(zos, sequences); LOG.info("Export phases: enumeration {} s, component table {} s, records {} s (chunks {} s, assembly {} s), reference table {} s", @@ -204,6 +243,11 @@ public EntityCountSummary compute() { zos.finish(); } } catch (Throwable e) { + if (causedByCancellation(e)) { + // A cancel can land inside a chunk's subtask, where the scope reports it wrapped. + LOG.info("Export cancelled"); + throw e instanceof CancellationException cancelled ? cancelled : new CancellationException("Export cancelled"); + } LOG.error("Caught " + e + " while Exporting Entities"); if (!(e instanceof RuntimeException rx && rx.getCause() instanceof InterruptedException)) { AlertStreams.dispatchToRoot(e); @@ -393,7 +437,9 @@ private List writeRecordEntries(ZipOutputStream zos, Buckets bucke final int chunkIndex = c; final long[] slice = Arrays.copyOfRange(nids, Math.min(nids.length, c * CHUNK_RECORDS), Math.min(nids.length, (c + 1) * CHUNK_RECORDS)); - permits.acquire(); + if (exportCancelled() || !StructuredScopes.acquireUnlessCancelled(permits, scope)) { + break; // cancelled, or a chunk failed: join() below reports a failure + } scope.fork(() -> { try { chunks[bucketIndex][chunkIndex] = writeChunk(bucketIndex + 1, chunkIndex, slice, spoolDir, sequences); @@ -406,6 +452,8 @@ private List writeRecordEntries(ZipOutputStream zos, Buckets bucke } scope.join(); } + // A cancelled run stopped forking, so some chunks were never written. + throwIfCancelled(); long chunked = System.nanoTime(); chunkNanos.set(chunked - forked); List result = new ArrayList<>(order.size()); @@ -498,6 +546,7 @@ private long writeRecords(long[] nids, OutputStream out) throws IOException { }); long count = 0; for (int i = 0; i < nids.length; i++) { + throwIfCancelled(); if (records[i] == null) { LOG.warn("No entity for nid {} at export time; its table entry stays, its record is absent", nids[i]); completedUnitOfWork(); @@ -505,16 +554,19 @@ private long writeRecords(long[] nids, OutputStream out) throws IOException { } Entity entity = EntityRecordFactory.make(records[i]); try { - if (entity instanceof StampEntity stampEntity) { - moduleNids.add(stampEntity.moduleNid()); - authorNids.add(stampEntity.authorNid()); - } Entity written = entity instanceof StampRecord stampRecord ? stampRecord.withoutSupersededUncommittedVersions() : entity; TinkarMsg pbTinkarMsg = entityTransformer.transform(written); pbTinkarMsg.writeDelimitedTo(out); count++; + if (entity instanceof StampEntity stampEntity) { + // Listed for the manifest only once the stamp is written: a skipped one + // (below) can name a module or author with no public id, and the manifest + // would then fail the whole export. + moduleNids.add(stampEntity.moduleNid()); + authorNids.add(stampEntity.authorNid()); + } } catch (RuntimeException e) { LOG.error("DANGLING REFERENCE — skipping " + entity.getClass().getSimpleName() + " nid=" + entity.nid() diff --git a/entity/src/main/java/dev/ikm/tinkar/entity/load/LoadEntitiesFromProtobufFile.java b/entity/src/main/java/dev/ikm/tinkar/entity/load/LoadEntitiesFromProtobufFile.java index 1a578ca0f..06c8d00a0 100644 --- a/entity/src/main/java/dev/ikm/tinkar/entity/load/LoadEntitiesFromProtobufFile.java +++ b/entity/src/main/java/dev/ikm/tinkar/entity/load/LoadEntitiesFromProtobufFile.java @@ -103,6 +103,7 @@ public class LoadEntitiesFromProtobufFile extends TrackingCallable 0) { // zis.available returns 1 until AFTER EOF has been reached TinkarMsg pbTinkarMsg = TinkarMsg.parseDelimitedFrom(zis); - permits.acquire(); + if (!StructuredScopes.acquireUnlessCancelled(permits, scope)) { + break; // a subtask failed; join() below reports it + } scope.fork(() -> { try { ScopedValue.where(SCOPED_TINKAR_MSG, pbTinkarMsg).call(() -> { @@ -438,7 +441,9 @@ private void importRecords(InputStream in, Runnable tick) throws Exception { TinkarMsg pbTinkarMsg; while ((pbTinkarMsg = TinkarMsg.parseDelimitedFrom(in)) != null) { TinkarMsg record = pbTinkarMsg; - permits.acquire(); + if (!StructuredScopes.acquireUnlessCancelled(permits, scope)) { + break; // a subtask failed; join() below reports it + } scope.fork(() -> ScopedValue.where(SCOPED_TINKAR_MSG, record).call(() -> { try { entityTransformer.transform(record, entityConsumer, (stampEntity) -> { @@ -526,7 +531,9 @@ private EntityCountSummary computeFormat3(long expectedImports, List readers = StructuredScopes.open()) { Semaphore readerPermits = new Semaphore(Math.max(1, Math.min(Math.max(1, rest.size()), Runtime.getRuntime().availableProcessors() / 2))); for (ChangeSetFormat.RecordEntry entry : rest) { - readerPermits.acquire(); + if (!StructuredScopes.acquireUnlessCancelled(readerPermits, readers)) { + break; // a reader failed; join() below reports it + } readers.fork(() -> { try { importEntry(zip, entry, registration, reading, finishedBytes, tick); diff --git a/integration/src/test/java/dev/ikm/tinkar/integration/builder/ExportSkipsStampWithLostAuthorIT.java b/integration/src/test/java/dev/ikm/tinkar/integration/builder/ExportSkipsStampWithLostAuthorIT.java new file mode 100644 index 000000000..bc94e0799 --- /dev/null +++ b/integration/src/test/java/dev/ikm/tinkar/integration/builder/ExportSkipsStampWithLostAuthorIT.java @@ -0,0 +1,90 @@ +package dev.ikm.tinkar.integration.builder; + +import dev.ikm.tinkar.common.id.PublicIds; +import dev.ikm.tinkar.common.service.EntityCountSummary; +import dev.ikm.tinkar.common.service.PrimitiveData; +import dev.ikm.tinkar.entity.EntityService; +import dev.ikm.tinkar.entity.RecordListBuilder; +import dev.ikm.tinkar.entity.StampRecord; +import dev.ikm.tinkar.entity.StampVersionRecord; +import dev.ikm.tinkar.entity.aggregator.TemporalEntityAggregator; +import dev.ikm.tinkar.entity.builder.ActiveStamp; +import dev.ikm.tinkar.entity.builder.KnowledgeSet; +import dev.ikm.tinkar.entity.builder.Stamp; +import dev.ikm.tinkar.entity.export.ExportEntitiesToProtobufFile; +import dev.ikm.tinkar.integration.helper.DataStore; +import dev.ikm.tinkar.integration.helper.TestHelper; +import dev.ikm.tinkar.terms.KernelTerm; +import dev.ikm.tinkar.terms.State; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInstance; + +import java.nio.file.Files; +import java.nio.file.Path; +import java.time.Instant; +import java.util.UUID; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +/** + * A stamp whose author the store can no longer name is skipped by an export, which still + * completes. + * + *

A store can hold such a stamp: one written in a session whose author never had an entity, + * by a store that then lost the author's UUID on restart. The export skips the stamp as a + * dangling reference — and used to fail anyway, at the very end, because it had already listed + * the stamp's author for the manifest before finding that the stamp could not be written. + */ +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +class ExportSkipsStampWithLostAuthorIT { + + @BeforeAll + static void beforeAll() { + TestHelper.startDataBase(DataStore.EPHEMERAL_STORE); + } + + @AfterAll + static void afterAll() { + TestHelper.stopDatabase(); + } + + @Test + void anExportSkipsTheStampAndCompletes() throws Exception { + KnowledgeSet set = KnowledgeSet.of("88888888-8888-5888-9888-888888888888"); + ActiveStamp stamp = Stamp.active("2026-07-03T00:00:00Z", + set.conceptRef("Lost probe author (Probe)"), + set.conceptRef("Lost probe module (Probe)"), + KernelTerm.DEVELOPMENT_PATH); + set.concept("Lost probe author (Probe)").at(stamp).synonym("Lost probe author"); + set.concept("Lost probe module (Probe)").at(stamp).synonym("Lost probe module"); + set.write(); + + // A nid well past any this store has minted, so it has no public id. + long lostAuthorNid = PrimitiveData.nid(PublicIds.of(UUID.randomUUID())) + 100_000; + assertThrows(IllegalStateException.class, () -> PrimitiveData.publicId(lostAuthorNid)); + EntityService.get().putEntity(stampWithAuthor(lostAuthorNid, + Instant.parse("2026-07-04T00:00:00Z").toEpochMilli())); + + Path out = Path.of(System.getProperty("user.dir")).resolve("target").resolve("export-lost-author.zip"); + Files.deleteIfExists(out); + EntityCountSummary summary = new ExportEntitiesToProtobufFile(out.toFile(), + new TemporalEntityAggregator(0L, Long.MAX_VALUE)).compute(); + + assertEquals(2, summary.conceptCount()); + assertEquals(1, summary.stampCount(), "the set's stamp, without the one whose author is lost"); + } + + private static StampRecord stampWithAuthor(long authorNid, long time) { + UUID stampUuid = UUID.randomUUID(); + RecordListBuilder versions = RecordListBuilder.make(); + StampRecord stamp = new StampRecord(stampUuid.getMostSignificantBits(), + stampUuid.getLeastSignificantBits(), null, PrimitiveData.nid(PublicIds.of(stampUuid)), versions); + versions.add(new StampVersionRecord(stamp, State.ACTIVE.nid(), time, authorNid, + KernelTerm.PRIMORDIAL_MODULE.nid(), KernelTerm.DEVELOPMENT_PATH.nid())); + versions.build(); + return stamp; + } +} diff --git a/integration/src/test/java/dev/ikm/tinkar/integration/changeSet/MultiPassImportFailureIT.java b/integration/src/test/java/dev/ikm/tinkar/integration/changeSet/MultiPassImportFailureIT.java new file mode 100644 index 000000000..85527faea --- /dev/null +++ b/integration/src/test/java/dev/ikm/tinkar/integration/changeSet/MultiPassImportFailureIT.java @@ -0,0 +1,64 @@ +package dev.ikm.tinkar.integration.changeSet; + +import dev.ikm.tinkar.entity.load.LoadEntitiesFromProtobufFile; +import dev.ikm.tinkar.fixtures.NewEphemeralKeyValueProvider; +import dev.ikm.tinkar.schema.TinkarMsg; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.api.io.TempDir; + +import java.io.File; +import java.io.FileOutputStream; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Path; +import java.time.Duration; +import java.util.zip.ZipEntry; +import java.util.zip.ZipOutputStream; + +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; + +/** + * A multi-pass import that hits a bad record must fail, not hang. + * + *

The reader limits in-flight subtasks with a semaphore, each subtask releasing its permit when + * it finishes. The first failure cancels the scope, and a cancelled scope does not run the subtasks + * forked after it — so with more records left than permits, those permits never come back and the + * reader used to block in {@code acquire()} forever. Seen for real migrating a 6-bit dataset to the + * 8-bit nid layout, where an out-of-range element sequence left the import stuck on + * "Starting multi-pass import". + */ +@ExtendWith(NewEphemeralKeyValueProvider.class) +class MultiPassImportFailureIT { + + /** Comfortably more records than the reader's permits (8 per core). */ + private static final int RECORD_COUNT = Runtime.getRuntime().availableProcessors() * 8 * 20; + + @TempDir + Path tempDir; + + @Test + void aFailingRecordFailsTheImportInsteadOfHanging() throws IOException { + File changeSet = writeChangeSetOfEmptyRecords(tempDir.resolve("bad-changeset.zip").toFile()); + LoadEntitiesFromProtobufFile loader = new LoadEntitiesFromProtobufFile(changeSet, true); + + // A record with no value set fails in pass 1 ("Tinkar message value not set"). + assertTimeoutPreemptively(Duration.ofSeconds(60), () -> assertThrows(RuntimeException.class, loader::compute)); + } + + private static File writeChangeSetOfEmptyRecords(File file) throws IOException { + try (ZipOutputStream zos = new ZipOutputStream(new FileOutputStream(file))) { + zos.putNextEntry(new ZipEntry("Entities")); + for (int i = 0; i < RECORD_COUNT; i++) { + TinkarMsg.getDefaultInstance().writeDelimitedTo(zos); + } + zos.closeEntry(); + zos.putNextEntry(new ZipEntry("META-INF/MANIFEST.MF")); + zos.write(("Manifest-Version: 1.0\nTotal-Count: " + RECORD_COUNT + "\n\n") + .getBytes(StandardCharsets.UTF_8)); + zos.closeEntry(); + } + return file; + } +} diff --git a/integration/src/test/java/dev/ikm/tinkar/integration/diagnostic/ExceptionMessageGateIT.java b/integration/src/test/java/dev/ikm/tinkar/integration/diagnostic/ExceptionMessageGateIT.java index 3cc7a1aa9..1c75079a0 100644 --- a/integration/src/test/java/dev/ikm/tinkar/integration/diagnostic/ExceptionMessageGateIT.java +++ b/integration/src/test/java/dev/ikm/tinkar/integration/diagnostic/ExceptionMessageGateIT.java @@ -52,7 +52,7 @@ * does not see an entity, a version, a vertex, or a tree joined into a message with its own * {@code toString()}; {@code EntityText} and {@code DiTreeText} are the forms for those. * - *

Six messages are about the nid itself and are allowed ({@link #KNOWN}). The gate fails + *

Seven messages are about the nid itself and are allowed ({@link #KNOWN}). The gate fails * when their number changes in either direction, so the list stays true. */ class ExceptionMessageGateIT { diff --git a/integration/src/test/java/dev/ikm/tinkar/integration/provider/spinedarray/ExportCancellationIT.java b/integration/src/test/java/dev/ikm/tinkar/integration/provider/spinedarray/ExportCancellationIT.java new file mode 100644 index 000000000..2eb7f69db --- /dev/null +++ b/integration/src/test/java/dev/ikm/tinkar/integration/provider/spinedarray/ExportCancellationIT.java @@ -0,0 +1,64 @@ +package dev.ikm.tinkar.integration.provider.spinedarray; + +import dev.ikm.tinkar.common.util.io.FileUtil; +import dev.ikm.tinkar.entity.export.ExportEntitiesToProtobufFile; +import dev.ikm.tinkar.entity.load.LoadEntitiesFromProtobufFile; +import dev.ikm.tinkar.fixtures.TestConstants; +import dev.ikm.tinkar.integration.helper.DataStore; +import dev.ikm.tinkar.integration.helper.TestHelper; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +import java.io.File; +import java.time.Duration; +import java.util.concurrent.CancellationException; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; + +/** + * An export stops when the caller that started it cancels, wherever it has got to, and says it + * was cancelled rather than that it failed. + * + *

The service runs exports as jobs a user can cancel, and tracks the cancel with a handle of + * its own, so it asks through {@link ExportEntitiesToProtobufFile#cancelWhen}. + */ +class ExportCancellationIT { + + private static final File DATASTORE_ROOT = + TestConstants.createFilePathInTargetFromClassName.apply(ExportCancellationIT.class); + private static final File EXPORT_FILE = new File(DATASTORE_ROOT.getParentFile(), "export-cancellation.zip"); + + @BeforeAll + static void beforeAll() { + FileUtil.recursiveDelete(DATASTORE_ROOT); + TestHelper.startDataBase(DataStore.SPINED_ARRAY_STORE, DATASTORE_ROOT); + new LoadEntitiesFromProtobufFile(TestConstants.PB_STARTER_DATA_REASONED).compute(); + } + + @AfterAll + static void afterAll() { + TestHelper.stopDatabase(); + } + + @Test + void anExportCancelledBeforeItStartsStopsAtTheFirstCheck() { + ExportEntitiesToProtobufFile export = new ExportEntitiesToProtobufFile(EXPORT_FILE); + export.cancelWhen(() -> true); + assertTimeoutPreemptively(Duration.ofSeconds(60), + () -> assertThrows(CancellationException.class, export::compute)); + } + + @Test + void anExportCancelledWhileWritingRecordsStops() { + // Asked once per record as the chunks are written, and a few times between phases: + // cancelled partway through the records. + AtomicInteger asked = new AtomicInteger(); + ExportEntitiesToProtobufFile export = new ExportEntitiesToProtobufFile(EXPORT_FILE); + export.cancelWhen(() -> asked.incrementAndGet() > 500); + assertTimeoutPreemptively(Duration.ofSeconds(60), + () -> assertThrows(CancellationException.class, export::compute)); + } +} diff --git a/integration/src/test/java/dev/ikm/tinkar/integration/provider/spinedarray/SpinedArrayAbsentComponentIdentityIT.java b/integration/src/test/java/dev/ikm/tinkar/integration/provider/spinedarray/SpinedArrayAbsentComponentIdentityIT.java new file mode 100644 index 000000000..fc88a9a55 --- /dev/null +++ b/integration/src/test/java/dev/ikm/tinkar/integration/provider/spinedarray/SpinedArrayAbsentComponentIdentityIT.java @@ -0,0 +1,94 @@ +package dev.ikm.tinkar.integration.provider.spinedarray; + +import dev.ikm.tinkar.common.id.PublicId; +import dev.ikm.tinkar.common.id.PublicIds; +import dev.ikm.tinkar.common.service.PrimitiveData; +import dev.ikm.tinkar.common.util.io.FileUtil; +import dev.ikm.tinkar.fixtures.TestConstants; +import dev.ikm.tinkar.integration.helper.DataStore; +import dev.ikm.tinkar.integration.helper.TestHelper; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +import java.io.File; + +import java.util.UUID; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; + +/** + * A component that is referenced but has no entity still has an identity a SpinedArray store can + * report. + * + *

Importing a time-range changeset does this: a navigation semantic in it can list a child that + * changed outside the range and so is not in the file. The child gets a nid from its UUID with no + * entity behind it. Converting the semantic back to public ids — for the gRPC service, or another + * export — asks the store which component that nid is, and used to fail with "does not maintain a + * reverse (nid to public id) identity map". + */ +class SpinedArrayAbsentComponentIdentityIT { + + private static final File DATASTORE_ROOT = + TestConstants.createFilePathInTargetFromClassName.apply(SpinedArrayAbsentComponentIdentityIT.class); + + @BeforeAll + static void startEmptySpinedArrayStore() { + FileUtil.recursiveDelete(DATASTORE_ROOT); + TestHelper.startDataBase(DataStore.SPINED_ARRAY_STORE, DATASTORE_ROOT); + } + + @AfterAll + static void stopStore() { + TestHelper.stopDatabase(); + } + + @Test + void aReferencedButAbsentComponentResolvesToTheIdItWasMintedFrom() { + PublicId absent = PublicIds.of(UUID.randomUUID()); + long nid = PrimitiveData.nid(absent); + + assertNull(dev.ikm.tinkar.common.service.internal.EntityStore.current().getBytes(nid), "the component should have no entity"); + assertEquals(absent, PrimitiveData.publicId(nid)); + // Asked again, as every conversion of the referring semantic does. + assertEquals(absent, PrimitiveData.publicId(nid)); + } + + @Test + void aSaveRecordsTheIdentity_soItSurvivesARestart() throws Exception { + // The store rebuilds its UUID map from entity bytes when it opens, so a nid with no entity + // used to come back with no identity at all. The save now writes it down; loading it is + // SpinedArrayAbsentIdentityReloadIT, since this test harness cannot reopen a store mid-class. + PublicId absent = PublicIds.of(UUID.randomUUID()); + long nid = PrimitiveData.nid(absent); + // Through the data controller: PrimitiveData.save() does not reach the store. + dev.ikm.tinkar.common.service.ServiceLifecycleManager.get() + .getServicesForGroup(dev.ikm.tinkar.common.service.ServiceExclusionGroup.DATA_PROVIDER) + .stream() + .filter(dev.ikm.tinkar.common.service.DataServiceController.class::isInstance) + .map(service -> (dev.ikm.tinkar.common.service.DataServiceController) service) + .forEach(dev.ikm.tinkar.common.service.DataServiceController::save); + + String saved = java.nio.file.Files.readString( + findFile(DATASTORE_ROOT, "absentIdentities.txt").toPath()); + org.junit.jupiter.api.Assertions.assertTrue( + saved.contains(nid + " " + absent.asUuidArray()[0]), "saved: " + saved); + } + + private static File findFile(File root, String name) throws java.io.IOException { + try (var paths = java.nio.file.Files.walk(root.toPath())) { + return paths.filter(p -> p.getFileName().toString().equals(name)).findFirst() + .orElseThrow(() -> new AssertionError(name + " not written under " + root)).toFile(); + } + } + + @Test + void everyUuidTheNidWasMintedFromIsReported() { + PublicId absent = PublicIds.of(UUID.randomUUID(), UUID.randomUUID()); + long nid = PrimitiveData.nid(absent); + + assertEquals(java.util.Set.copyOf(absent.asUuidList().castToList()), + java.util.Set.copyOf(PrimitiveData.publicId(nid).asUuidList().castToList())); + } +} diff --git a/integration/src/test/java/dev/ikm/tinkar/integration/provider/spinedarray/SpinedArrayAbsentIdentityReloadIT.java b/integration/src/test/java/dev/ikm/tinkar/integration/provider/spinedarray/SpinedArrayAbsentIdentityReloadIT.java new file mode 100644 index 000000000..a1174cb44 --- /dev/null +++ b/integration/src/test/java/dev/ikm/tinkar/integration/provider/spinedarray/SpinedArrayAbsentIdentityReloadIT.java @@ -0,0 +1,52 @@ +package dev.ikm.tinkar.integration.provider.spinedarray; + +import dev.ikm.tinkar.common.id.PublicId; +import dev.ikm.tinkar.common.id.PublicIds; +import dev.ikm.tinkar.common.service.PrimitiveData; +import dev.ikm.tinkar.common.util.io.FileUtil; +import dev.ikm.tinkar.fixtures.TestConstants; +import dev.ikm.tinkar.integration.helper.DataStore; +import dev.ikm.tinkar.integration.helper.TestHelper; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +import java.io.File; +import java.nio.file.Files; +import java.util.UUID; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * A SpinedArray store that opens with an {@code absentIdentities.txt} — written by an earlier + * session's save — knows the components it names again: their UUIDs resolve to the same nids, so + * references to them still hold, and asking which component a nid is still has an answer. + */ +class SpinedArrayAbsentIdentityReloadIT { + + private static final File DATASTORE_ROOT = + TestConstants.createFilePathInTargetFromClassName.apply(SpinedArrayAbsentIdentityReloadIT.class); + private static final PublicId ABSENT = PublicIds.of(UUID.randomUUID()); + /** A nid well below where a fresh store starts minting, as an earlier session would have used. */ + private static final int NID = Integer.MIN_VALUE + 7; + + @BeforeAll + static void openAStoreWithASavedAbsentIdentity() throws Exception { + FileUtil.recursiveDelete(DATASTORE_ROOT); + Files.createDirectories(DATASTORE_ROOT.toPath()); + Files.writeString(DATASTORE_ROOT.toPath().resolve("absentIdentities.txt"), + NID + " " + ABSENT.asUuidArray()[0] + "\n"); + TestHelper.startDataBase(DataStore.SPINED_ARRAY_STORE, DATASTORE_ROOT); + } + + @AfterAll + static void stopStore() { + TestHelper.stopDatabase(); + } + + @Test + void theSavedIdentityIsKnownAgain() { + assertEquals(ABSENT, PrimitiveData.publicId(NID)); + assertEquals(NID, PrimitiveData.nid(ABSENT)); + } +} diff --git a/pom.xml b/pom.xml index 893acd514..c64428087 100644 --- a/pom.xml +++ b/pom.xml @@ -11,7 +11,7 @@ dev.ikm.tinkar tinkar-core - 1.127.7-SNAPSHOT + 1.127.7-changeset_import_export-SNAPSHOT pom IKM tinkar core diff --git a/provider/data-spinedarray-provider/src/main/java/dev/ikm/tinkar/provider/spinedarray/SpinedArrayProvider.java b/provider/data-spinedarray-provider/src/main/java/dev/ikm/tinkar/provider/spinedarray/SpinedArrayProvider.java index 3c44dd2c1..70286d014 100644 --- a/provider/data-spinedarray-provider/src/main/java/dev/ikm/tinkar/provider/spinedarray/SpinedArrayProvider.java +++ b/provider/data-spinedarray-provider/src/main/java/dev/ikm/tinkar/provider/spinedarray/SpinedArrayProvider.java @@ -22,6 +22,7 @@ import dev.ikm.tinkar.common.id.Nid; import dev.ikm.tinkar.common.service.SequentialNids; import dev.ikm.tinkar.common.service.internal.EntityStore; +import dev.ikm.tinkar.common.id.PublicIds; import dev.ikm.tinkar.common.util.SetOnce; import dev.ikm.tinkar.collection.SpinedByteArrayMap; import dev.ikm.tinkar.collection.SpinedIntLongArrayMap; @@ -134,6 +135,16 @@ public static SpinedArrayProvider get() { final File nidToByteArrayMapDirectory; final File nidToCitingComponentNidMapDirectory; final File nextNidKeyFile; + + /** + * The UUIDs of nids that have no entity — components referenced but not present, such as a + * child listed by an imported navigation semantic whose own record was not in the changeset. + * + *

The UUID map is rebuilt from entity bytes at startup, so without this file such a nid + * would lose its identity on restart: nothing could say which component it is, and importing + * that component later would give it a second nid, leaving the reference dangling for good. + */ + final File absentIdentitiesFile; final SetOnce searchService = new SetOnce<>(); private volatile boolean loadPhase = false; final String name; @@ -165,6 +176,7 @@ private SpinedArrayProvider(boolean ephemeral) throws IOException, ExecutionExce this.nidToByteArrayMapDirectory = null; this.nidToCitingComponentNidMapDirectory = null; this.nextNidKeyFile = null; + this.absentIdentitiesFile = null; this.entityToBytesMap = new SpinedByteArrayMap(new ByteArrayNoStore()); this.nidToCitingComponentsNidMap = new SpinedIntLongArrayMap(new IntLongArrayNoStore()); LOG.info("Datastore: ephemeral, held in memory, nothing on disk"); @@ -184,6 +196,7 @@ private SpinedArrayProvider(boolean ephemeral) throws IOException, ExecutionExce this.nidToCitingComponentNidMapDirectory = new File(configuredRoot, "nidToCitingComponentNidMap"); this.nidToCitingComponentNidMapDirectory.mkdirs(); this.nextNidKeyFile = new File(configuredRoot, "nextNidKeyFile"); + this.absentIdentitiesFile = new File(configuredRoot, "absentIdentities.txt"); this.entityToBytesMap = new SpinedByteArrayMap(new ByteArrayFileStore(nidToByteArrayMapDirectory)); this.nidToCitingComponentsNidMap = new SpinedIntLongArrayMap(new IntLongArrayFileStore(nidToCitingComponentNidMapDirectory)); @@ -193,6 +206,10 @@ private SpinedArrayProvider(boolean ephemeral) throws IOException, ExecutionExce nextNid.set(Integer.valueOf(nextNidString)); } } + // Nids minted before this opening were either given entities or recorded in + // absentIdentitiesFile by an earlier save; only newer ones need checking. + checkedBelowNid = nextNid.get(); + loadAbsentIdentities(); LOG.info("Submitting UUID loading task to thread pool..."); try { TinkExecutor.threadPool().submit(() -> { @@ -307,6 +324,7 @@ public void save() { Files.writeString(this.nextNidKeyFile.toPath(), Integer.toString(nextNid.get())); this.entityToBytesMap.write(); this.nidToCitingComponentsNidMap.write(); + saveAbsentIdentities(); } catch (Exception e) { LOG.error("Error saving SpinedArrayProvider", e); } finally { @@ -315,6 +333,74 @@ public void save() { } } + /** + * UUIDs of nids that have no entity: loaded from {@link #absentIdentitiesFile}, and found on + * each save among the nids minted since the previous one. Small — such nids are rare. + */ + private final java.util.concurrent.ConcurrentHashMap> absentUuids = + new java.util.concurrent.ConcurrentHashMap<>(); + + /** Nids below this have been checked for missing entities by a save. Set when the store opens. */ + private volatile int checkedBelowNid = Integer.MIN_VALUE; + + private void loadAbsentIdentities() throws IOException { + if (absentIdentitiesFile == null || !absentIdentitiesFile.exists()) { + return; + } + int loaded = 0; + for (String line : Files.readAllLines(absentIdentitiesFile.toPath())) { + String[] parts = line.trim().split("\\s+"); + if (parts.length < 2) { + continue; + } + int nid = Integer.parseInt(parts[0]); + List uuids = new java.util.concurrent.CopyOnWriteArrayList<>(); + for (int i = 1; i < parts.length; i++) { + UUID uuid = UUID.fromString(parts[i]); + uuids.add(uuid); + uuidToNidMap.putIfAbsent(uuid, nid); + } + absentUuids.put(nid, uuids); + loaded++; + } + LOG.info("Loaded {} identities of referenced-but-absent components", loaded); + } + + /** + * Writes the UUIDs of every nid still without an entity. Checks only the nids minted since the + * last save — nids are handed out in sequence — and finds the UUIDs of any new ones in a single + * pass over the identity map, so a save costs little unless something was actually left absent. + */ + private void saveAbsentIdentities() throws IOException { + absentUuids.keySet().removeIf(nid -> entityToBytesMap.get(nid) != null); + int mintedUpTo = nextNid.get(); + java.util.Set newlyAbsent = new java.util.HashSet<>(); + for (int nid = checkedBelowNid; nid < mintedUpTo; nid++) { + if (!absentUuids.containsKey(nid) && entityToBytesMap.get(nid) == null) { + newlyAbsent.add(nid); + } + } + if (!newlyAbsent.isEmpty()) { + uuidToNidMap.forEach((uuid, nid) -> { + if (newlyAbsent.contains(nid)) { + absentUuids.computeIfAbsent(nid, key -> new java.util.concurrent.CopyOnWriteArrayList<>()).add(uuid); + } + }); + } + checkedBelowNid = mintedUpTo; + if (absentUuids.isEmpty()) { + Files.deleteIfExists(absentIdentitiesFile.toPath()); + return; + } + StringBuilder lines = new StringBuilder(); + absentUuids.forEach((nid, uuids) -> { + lines.append(nid); + uuids.forEach(uuid -> lines.append(' ').append(uuid)); + lines.append('\n'); + }); + Files.writeString(absentIdentitiesFile.toPath(), lines); + } + /** * The nid of a public id, through the sequential providers' shared resolution * ({@link SequentialNids}): one UUID is looked up or minted; several resolve by the least @@ -362,6 +448,10 @@ public boolean hasPublicId(PublicId publicId) { * The reverse of {@link #nidForUuids}, read from the identity map, which is kept one way: * a scan of it, correctness over speed, for a referenced component no entity was written * for. The entity layer asks only after it found no entity for the nid. + * + *

The identity map is rebuilt from entities when the store opens, so a referenced + * component with no entity would lose its identity on restart. Such identities are saved to + * {@link #absentIdentitiesFile} and read here first. */ @Override public PublicId publicIdForNid(long nid) { @@ -371,6 +461,10 @@ public PublicId publicIdForNid(long nid) { throw new RuntimeException(e); } int narrowed = Nid.narrowChecked(nid); + List known = absentUuids.get(narrowed); + if (known != null && !known.isEmpty()) { + return PublicIds.of(known); + } List uuids = new ArrayList<>(); uuidToNidMap.forEach((uuid, mappedNid) -> { if (mappedNid == narrowed) { diff --git a/reasoner/reasoner-elk-snomed-test/pom.xml b/reasoner/reasoner-elk-snomed-test/pom.xml index 4bdfaaf9b..a570196b1 100644 --- a/reasoner/reasoner-elk-snomed-test/pom.xml +++ b/reasoner/reasoner-elk-snomed-test/pom.xml @@ -5,7 +5,7 @@ dev.ikm.tinkar reasoner - 1.127.7-SNAPSHOT + 1.127.7-changeset_import_export-SNAPSHOT reasoner-elk-snomed-test diff --git a/reasoner/reasoner-elk-snomed/src/main/java/dev/ikm/tinkar/reasoner/elksnomed/ElkSnomedDataBuilder.java b/reasoner/reasoner-elk-snomed/src/main/java/dev/ikm/tinkar/reasoner/elksnomed/ElkSnomedDataBuilder.java index bdd387d7f..b7e20e9f7 100644 --- a/reasoner/reasoner-elk-snomed/src/main/java/dev/ikm/tinkar/reasoner/elksnomed/ElkSnomedDataBuilder.java +++ b/reasoner/reasoner-elk-snomed/src/main/java/dev/ikm/tinkar/reasoner/elksnomed/ElkSnomedDataBuilder.java @@ -17,6 +17,7 @@ import dev.ikm.tinkar.terms.KernelTerm; import java.math.BigDecimal; +import java.util.concurrent.CancellationException; import java.util.concurrent.atomic.AtomicInteger; import org.eclipse.collections.api.list.ImmutableList; @@ -83,6 +84,10 @@ private void updateProgress(int count, int total) { progressUpdater.updateProgress(count, total); } + private boolean isCancelled() { + return progressUpdater != null && progressUpdater.isCancelled(); + } + private int computeTotalCount() { // Counted without reading them: the build reads them after (IKE-Network/ike-issues#1249). return EntityService.get().countSemanticsOfPattern(statedAxiomPattern.nid()); @@ -99,6 +104,11 @@ public void build() throws Exception { AtomicInteger ex_cnt = new AtomicInteger(); viewCalculator.forEachSemanticVersionOfPatternParallel(logicCoordinate.statedAxiomsPatternNid(), (semanticEntityVersion, _) -> { + // The parallel traversal cannot be stopped from inside, so once cancelled each + // remaining axiom is skipped instead — it is processing them that takes the time. + if (isCancelled()) { + return; + } try { if (semanticEntityVersion.active()) { processDefinition(semanticEntityVersion); @@ -123,6 +133,9 @@ public void build() throws Exception { } } }); + if (isCancelled()) { + throw new CancellationException("Reasoner cancelled while extracting stated axioms"); + } buildRoleConcepts(); data.initializeReasonerConceptSet(); LOG.info("Reasoner concept set size: {}, concepts loaded: {}", diff --git a/reasoner/reasoner-elk-snomed/src/main/java/dev/ikm/tinkar/reasoner/elksnomed/ElkSnomedReasonerService.java b/reasoner/reasoner-elk-snomed/src/main/java/dev/ikm/tinkar/reasoner/elksnomed/ElkSnomedReasonerService.java index 4bdfa0c98..e4a9cb087 100644 --- a/reasoner/reasoner-elk-snomed/src/main/java/dev/ikm/tinkar/reasoner/elksnomed/ElkSnomedReasonerService.java +++ b/reasoner/reasoner-elk-snomed/src/main/java/dev/ikm/tinkar/reasoner/elksnomed/ElkSnomedReasonerService.java @@ -148,11 +148,18 @@ private CancelWatcher(TrackingCallable progressTracker, SnomedOntologyReasone return; } this.watcher = new Thread(() -> { + boolean announced = false; while (!stopped) { if (progressTracker.isCancelled()) { - LOG.info("Cancel requested — interrupting the reasoner"); + if (!announced) { + LOG.info("Cancel requested — interrupting the reasoner"); + announced = true; + } + // Every poll, not once: until computeInferences() has created ELK's + // reasoner, interrupt() has nothing to act on. A cancel that arrived + // before that — while data was still loading — would otherwise be dropped + // and the whole classification would run before the cancel took effect. reasoner.interrupt(); - return; } try { Thread.sleep(POLL_INTERVAL_MS);