78: Make JournalService a Hub-managed Nenjim process and integrate it into startup

This commit is contained in:
2026-09-02 16:46:51 +02:00
parent f5fc5bb3b5
commit 7b1f6d18eb
19 changed files with 858 additions and 254 deletions
@@ -29,6 +29,9 @@ import com.r35157.libs.solana.impl.cached.CachedSolanaBlockChain;
import com.r35157.libs.valuetypes.basic.MoneyAmount;
import com.r35157.nenjim.hubd.NenjimHub;
import com.r35157.nenjim.process.NenjimProcess;
import com.r35157.nenjim.service.journal.JournalService;
import com.r35157.nenjim.service.journal.JournalServiceManager;
import com.r35157.nenjim.service.journal.impl.ref.JournalServiceAssembly;
import com.r35157.nenjim.service.registry.NenjimRegistryService;
import com.r35157.nenjim.service.registry.NenjimRegistryServiceManager;
import com.r35157.nenjim.service.registry.impl.ref.NenjimRegistryAssembly;
@@ -47,6 +50,7 @@ import org.slf4j.LoggerFactory;
import java.math.BigDecimal;
import java.net.URI;
import java.net.http.HttpClient;
import java.nio.file.Path;
import java.time.Clock;
import java.time.Duration;
import java.util.ArrayList;
@@ -79,6 +83,7 @@ public final class NenjimHubImpl implements NenjimHub {
REGISTRY_SERVICE_ID,
registryService);
recordStartedProcess(REGISTRY_SERVICE_ID);
initializeAndStartJournal(registryServiceManager);
initializeAndRegisterComponents(
registryService,
registryServiceManager);
@@ -101,6 +106,22 @@ public final class NenjimHubImpl implements NenjimHub {
return List.copyOf(startedProcessIds);
}
private void initializeAndStartJournal(
NenjimRegistryServiceManager registryServiceManager)
throws Exception {
JournalServiceAssembly journalAssembly =
JournalServiceAssembly.create(JOURNAL_ROOT);
JournalService journalService = journalAssembly.journalService();
NenjimProcess journalProcess = journalAssembly.journalProcess();
journalServiceManager = journalAssembly.journalServiceManager();
registryServiceManager.registerComponent(
JOURNAL_SERVICE_ID,
journalService);
journalProcess.start();
recordStartedProcess(JOURNAL_SERVICE_ID);
}
private void initializeAndRegisterComponents(
NenjimRegistryService registryService,
NenjimRegistryServiceManager registryServiceManager) {
@@ -290,6 +311,10 @@ public final class NenjimHubImpl implements NenjimHub {
NenjimHubImpl.class);
private static final NenjimComponentId REGISTRY_SERVICE_ID =
new NenjimComponentId("nenjim.registry.service");
private static final NenjimComponentId JOURNAL_SERVICE_ID =
new NenjimComponentId("nenjim.journal.service");
private static final Path JOURNAL_ROOT =
Path.of("data/nenjim/journals");
private static final NenjimComponentId OBJECT_CACHE_ID =
new NenjimComponentId("nenjim.object-cache.default");
private static final NenjimComponentId CURRENCY_IDENTITY_SERVICE_ID =
@@ -337,6 +362,7 @@ public final class NenjimHubImpl implements NenjimHub {
private final AtomicBoolean startAttempted = new AtomicBoolean();
private final List<NenjimComponentId> startedProcessIds = new ArrayList<>();
private NenjimRegistryServiceManager registryServiceManager;
private JournalServiceManager journalServiceManager;
// Dependency components
private ObjectCache objectCache;
@@ -1,6 +1,7 @@
package com.r35157.nenjim.service.journal;
import com.r35157.libs.valuetypes.basic.SemanticVersion;
import com.r35157.nenjim.component.NenjimComponent;
import com.r35157.nenjim.service.journal.model.ArtifactRelease;
import com.r35157.nenjim.service.journal.model.Journal;
import com.r35157.nenjim.service.journal.valuetypes.ArtifactCoordinate;
@@ -10,8 +11,15 @@ import java.util.List;
import java.util.Optional;
import java.util.Set;
/** Filesystem-independent, read-only access to current and historical Journal knowledge. */
public interface JournalService {
/**
* Filesystem-independent, read-only access to current and historical Journal
* knowledge.
*
* <p>The service is a registrable Nenjim component, but exposes neither its
* process lifecycle nor its filesystem administration to ordinary
* consumers.</p>
*/
public interface JournalService extends NenjimComponent {
/** Lists known artifact coordinates in stable lexical order. */
Set<ArtifactCoordinate> artifactCoordinates();
@@ -1,22 +1,19 @@
package com.r35157.nenjim.service.journal;
import org.jetbrains.annotations.NotNull;
import java.io.IOException;
import java.nio.file.Path;
/** Lifecycle and administrative access for one filesystem-backed Journal service. */
/** Passive filesystem administration for one Hub-owned Journal service. */
public interface JournalServiceManager {
/** Starts the service and completes its initial Journal loading before returning. */
void start() throws IOException;
/** Stops service-owned maintenance without changing Journal files. */
void stop();
/** Returns the normalized configured Journal data root. */
@NotNull
Path journalRoot();
/** Returns the read-only service after successful initialization. */
JournalService journalService();
/** Explicitly reloads the configured runtime Journal files. */
/**
* Explicitly reloads the configured runtime Journal files after the
* Journal process has started successfully.
*/
void refresh() throws IOException;
}
@@ -0,0 +1,146 @@
package com.r35157.nenjim.service.journal.impl.ref;
import com.r35157.nenjim.service.journal.model.Journal;
import com.r35157.nenjim.service.journal.valuetypes.ArtifactCoordinate;
import com.r35157.nenjim.service.journal.valuetypes.JournalVersion;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Objects;
import java.util.stream.Stream;
/** Strict, read-only filesystem loading for one Journal service. */
final class JournalFilesystemLoader {
private static final String JOURNAL_SUFFIX = ".journal";
private static final String EXAMPLE_SUFFIX = ".journal.example";
private JournalFilesystemLoader() {
}
static void loadAll(
Path journalRoot,
JournalServiceImpl service) throws IOException {
validateRoot(journalRoot);
for (Path artifactDirectory : listEntries(journalRoot)) {
if (!Files.isDirectory(artifactDirectory)) {
throw new IOException(
"Unexpected entry in Journal root; expected an artifact directory: "
+ artifactDirectory);
}
loadArtifactDirectory(service, artifactDirectory);
}
}
private static void validateRoot(Path journalRoot) throws IOException {
if (!Files.exists(journalRoot)) {
throw new IOException(
"Configured Journal root does not exist: " + journalRoot);
}
if (!Files.isDirectory(journalRoot)) {
throw new IOException(
"Configured Journal root is not a directory: " + journalRoot);
}
if (!Files.isReadable(journalRoot)) {
throw new IOException(
"Configured Journal root is not readable: " + journalRoot);
}
}
private static void loadArtifactDirectory(
JournalServiceImpl service,
Path directory) throws IOException {
String directoryName = directory.getFileName().toString();
ArtifactCoordinate expectedCoordinate;
try {
expectedCoordinate = ArtifactCoordinate.parse(directoryName);
} catch (IllegalArgumentException exception) {
throw new IOException(
"Invalid artifact-coordinate directory name: " + directory,
exception);
}
if (!Files.isReadable(directory)) {
throw new IOException(
"Artifact Journal directory is not readable: " + directory);
}
ArrayList<JournalFile> runtimeFiles = new ArrayList<>();
for (Path entry : listEntries(directory)) {
if (!Files.isRegularFile(entry)) {
throw new IOException(
"Unexpected entry in artifact Journal directory: " + entry);
}
String filename = entry.getFileName().toString();
if (filename.endsWith(EXAMPLE_SUFFIX)) {
continue;
}
if (!filename.endsWith(JOURNAL_SUFFIX)) {
throw new IOException(
"Unexpected file in artifact Journal directory: " + entry);
}
if (!Files.isReadable(entry)) {
throw new IOException("Journal file is not readable: " + entry);
}
String versionText = filename.substring(
0,
filename.length() - JOURNAL_SUFFIX.length());
JournalVersion expectedVersion;
try {
expectedVersion = JournalVersion.parse(versionText);
} catch (IllegalArgumentException exception) {
throw new IOException(
"Invalid Journal filename; expected <journal-version>.journal: "
+ entry,
exception);
}
runtimeFiles.add(new JournalFile(entry, expectedVersion));
}
runtimeFiles.sort(Comparator.comparing(JournalFile::journalVersion));
for (JournalFile runtimeFile : runtimeFiles) {
String text = Files.readString(
runtimeFile.path,
StandardCharsets.UTF_8);
String source = runtimeFile.path.toString();
Journal parsed = service.parse(text, source);
if (!parsed.metadata().coordinate().equals(expectedCoordinate)) {
throw new IOException(
"Journal metadata coordinate "
+ parsed.metadata().coordinate()
+ " does not match directory "
+ expectedCoordinate + ": " + runtimeFile.path);
}
if (!parsed.journalVersion().equals(runtimeFile.journalVersion)) {
throw new IOException(
"JOURNAL_VERSION " + parsed.journalVersion()
+ " does not match filename version "
+ runtimeFile.journalVersion + ": "
+ runtimeFile.path);
}
service.ingest(text, source);
}
}
private static List<Path> listEntries(Path directory) throws IOException {
try (Stream<Path> entries = Files.list(directory)) {
return entries.sorted(Comparator.comparing(
path -> path.getFileName().toString())).toList();
} catch (SecurityException exception) {
throw new IOException(
"Cannot read Journal directory: " + directory,
exception);
}
}
private record JournalFile(Path path, JournalVersion journalVersion) {
private JournalFile {
Objects.requireNonNull(path, "path");
Objects.requireNonNull(journalVersion, "journalVersion");
}
}
}
@@ -0,0 +1,22 @@
package com.r35157.nenjim.service.journal.impl.ref;
import org.jetbrains.annotations.NotNull;
import java.io.IOException;
import java.nio.file.Path;
/**
* Internal administrative view of the active Journal service.
*
* <p>The same object is exposed publicly through the read-only
* {@code JournalService} interface. This interface must remain package-private:
* making it public would let ordinary consumers cast that service object to
* filesystem administration and would break the read-only service contract.
* It must never be exposed outside this reference implementation package.</p>
*/
interface JournalServiceAdmin {
@NotNull
Path journalRoot();
void refresh() throws IOException;
}
@@ -0,0 +1,51 @@
package com.r35157.nenjim.service.journal.impl.ref;
import com.r35157.nenjim.process.NenjimProcess;
import com.r35157.nenjim.service.journal.JournalService;
import com.r35157.nenjim.service.journal.JournalServiceManager;
import org.jetbrains.annotations.NotNull;
import java.nio.file.Path;
import java.util.Objects;
/**
* Implementation-only construction boundary for one Journal subsystem.
*
* <p>This helper owns no lifecycle and exposes neither parser, ingestion,
* chain state nor filesystem administration through the ordinary service
* view. A consumer holding only {@link JournalService} cannot use that
* reference to derive the manager bound to the active Journal service.</p>
*/
public final class JournalServiceAssembly {
private JournalServiceAssembly(@NotNull JournalServiceImpl service) {
journalService = service;
journalProcess = service;
journalServiceManager = new JournalServiceManagerImpl(service);
}
/** Creates mutually consistent public views for one configured Journal. */
public static @NotNull JournalServiceAssembly create(
@NotNull Path journalRoot) {
return new JournalServiceAssembly(new JournalServiceImpl(
Objects.requireNonNull(journalRoot, "journalRoot")));
}
/** Returns the ordinary read-only service view. */
public @NotNull JournalService journalService() {
return journalService;
}
/** Returns the process view of the same object as the service view. */
public @NotNull NenjimProcess journalProcess() {
return journalProcess;
}
/** Returns the distinct passive manager bound to this Journal service. */
public @NotNull JournalServiceManager journalServiceManager() {
return journalServiceManager;
}
private final JournalService journalService;
private final NenjimProcess journalProcess;
private final JournalServiceManager journalServiceManager;
}
@@ -1,6 +1,7 @@
package com.r35157.nenjim.service.journal.impl.ref;
import com.r35157.libs.valuetypes.basic.SemanticVersion;
import com.r35157.nenjim.process.NenjimProcess;
import com.r35157.nenjim.service.journal.JournalService;
import com.r35157.nenjim.service.journal.exception.InvalidJournalException;
import com.r35157.nenjim.service.journal.model.ArtifactRelease;
@@ -9,7 +10,9 @@ import com.r35157.nenjim.service.journal.valuetypes.ArtifactCoordinate;
import com.r35157.nenjim.service.journal.valuetypes.ContentDigest;
import com.r35157.nenjim.service.journal.valuetypes.JournalVersion;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Path;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
@@ -25,26 +28,68 @@ import java.util.Optional;
import java.util.Set;
import java.util.TreeMap;
import java.util.TreeSet;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* Filesystem-independent, thread-safe reference service for immutable per-artifact Journal chains.
* Thread-safe reference service and Hub-managed process for immutable
* per-artifact Journal chains.
*/
public final class JournalServiceImpl implements JournalService {
final class JournalServiceImpl implements
JournalService,
NenjimProcess,
JournalServiceAdmin {
private static final String DEFAULT_SOURCE = "<journal>";
private final Object stateLock = new Object();
private final Path journalRoot;
private final JournalTextParser parser;
private final Map<ArtifactCoordinate, NavigableMap<JournalVersion, StoredRevision>> chains = new HashMap<>();
private final Map<String, ArtifactCoordinate> digestOwners = new HashMap<>();
private final AtomicBoolean startAttempted = new AtomicBoolean();
private volatile boolean available;
JournalServiceImpl() {
this(new JournalTextParser());
JournalServiceImpl(Path journalRoot) {
this(journalRoot, new JournalTextParser());
}
JournalServiceImpl(JournalTextParser parser) {
JournalServiceImpl(Path journalRoot, JournalTextParser parser) {
this.journalRoot = Objects.requireNonNull(
journalRoot,
"journalRoot").toAbsolutePath().normalize();
this.parser = Objects.requireNonNull(parser, "parser");
}
/**
* Loads the complete initial filesystem worldview and publishes it only
* after every runtime Journal has been accepted.
*/
@Override
public void start() throws IOException {
if (!startAttempted.compareAndSet(false, true)) {
throw new IllegalStateException(
"Journal service start has already been attempted");
}
JournalServiceImpl candidate = new JournalServiceImpl(
journalRoot,
parser);
JournalFilesystemLoader.loadAll(journalRoot, candidate);
publishInitialState(candidate);
}
/** Returns the normalized configured Journal data root. */
@Override
public Path journalRoot() {
return journalRoot;
}
/** Explicitly reloads runtime Journal files after successful startup. */
@Override
public void refresh() throws IOException {
requireAvailable();
JournalFilesystemLoader.loadAll(journalRoot, this);
}
/** Parses and validates a complete snapshot without changing service state. */
Journal parse(String completeJournalText) {
return parse(completeJournalText, DEFAULT_SOURCE);
@@ -114,6 +159,7 @@ public final class JournalServiceImpl implements JournalService {
/** Lists known artifact coordinates in stable lexical order. */
@Override
public Set<ArtifactCoordinate> artifactCoordinates() {
requireAvailable();
synchronized (stateLock) {
TreeSet<ArtifactCoordinate> coordinates = new TreeSet<>(Comparator.comparing(ArtifactCoordinate::toString));
coordinates.addAll(chains.keySet());
@@ -124,6 +170,7 @@ public final class JournalServiceImpl implements JournalService {
/** Lists complete revisions in ascending Journal-version order. */
@Override
public List<Journal> revisions(ArtifactCoordinate coordinate) {
requireAvailable();
Objects.requireNonNull(coordinate, "coordinate");
synchronized (stateLock) {
NavigableMap<JournalVersion, StoredRevision> chain = chains.get(coordinate);
@@ -137,6 +184,7 @@ public final class JournalServiceImpl implements JournalService {
/** Returns the current complete worldview for an artifact. */
@Override
public Optional<Journal> latest(ArtifactCoordinate coordinate) {
requireAvailable();
Objects.requireNonNull(coordinate, "coordinate");
synchronized (stateLock) {
NavigableMap<JournalVersion, StoredRevision> chain = chains.get(coordinate);
@@ -147,6 +195,7 @@ public final class JournalServiceImpl implements JournalService {
/** Returns one exact historical complete worldview. */
@Override
public Optional<Journal> revision(ArtifactCoordinate coordinate, JournalVersion journalVersion) {
requireAvailable();
Objects.requireNonNull(coordinate, "coordinate");
Objects.requireNonNull(journalVersion, "journalVersion");
synchronized (stateLock) {
@@ -162,6 +211,7 @@ public final class JournalServiceImpl implements JournalService {
public Optional<Journal> revisionAtOrBefore(
ArtifactCoordinate coordinate,
JournalVersion journalVersion) {
requireAvailable();
Objects.requireNonNull(coordinate, "coordinate");
Objects.requireNonNull(journalVersion, "journalVersion");
synchronized (stateLock) {
@@ -178,6 +228,7 @@ public final class JournalServiceImpl implements JournalService {
ArtifactCoordinate coordinate,
JournalVersion journalVersion,
SemanticVersion releaseVersion) {
requireAvailable();
Objects.requireNonNull(releaseVersion, "releaseVersion");
return revision(coordinate, journalVersion).flatMap(journal -> journal.release(releaseVersion));
}
@@ -196,6 +247,23 @@ public final class JournalServiceImpl implements JournalService {
}
}
private void publishInitialState(JournalServiceImpl candidate) {
synchronized (candidate.stateLock) {
synchronized (stateLock) {
chains.putAll(candidate.chains);
digestOwners.putAll(candidate.digestOwners);
available = true;
}
}
}
private void requireAvailable() {
if (!available) {
throw new IllegalStateException(
"Journal service is unavailable until startup succeeds");
}
}
private InvalidJournalException invalidPredecessor(
String source,
ArtifactCoordinate candidateCoordinate,
@@ -1,164 +1,29 @@
package com.r35157.nenjim.service.journal.impl.ref;
import com.r35157.nenjim.service.journal.JournalService;
import com.r35157.nenjim.service.journal.JournalServiceManager;
import com.r35157.nenjim.service.journal.model.Journal;
import com.r35157.nenjim.service.journal.valuetypes.ArtifactCoordinate;
import com.r35157.nenjim.service.journal.valuetypes.JournalVersion;
import org.jetbrains.annotations.NotNull;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Objects;
import java.util.stream.Stream;
/** Filesystem-backed reference lifecycle manager for a read-only Journal service. */
/** Passive reference manager for one Hub-owned Journal service. */
public final class JournalServiceManagerImpl implements JournalServiceManager {
private static final String JOURNAL_SUFFIX = ".journal";
private static final String EXAMPLE_SUFFIX = ".journal.example";
private final Path journalRoot;
private JournalServiceImpl activeService;
public JournalServiceManagerImpl(Path journalRoot) {
this.journalRoot = Objects.requireNonNull(journalRoot, "journalRoot").toAbsolutePath().normalize();
JournalServiceManagerImpl(JournalServiceAdmin journalAdmin) {
this.journalAdmin = Objects.requireNonNull(
journalAdmin,
"journalAdmin");
}
@Override
public synchronized void start() throws IOException {
if (activeService != null) {
return;
}
JournalServiceImpl candidate = new JournalServiceImpl();
loadAll(candidate);
activeService = candidate;
}
@Override
public synchronized void stop() {
activeService = null;
}
@Override
public Path journalRoot() {
return journalRoot;
}
@Override
public synchronized JournalService journalService() {
return requireActiveService();
public @NotNull Path journalRoot() {
return journalAdmin.journalRoot();
}
@Override
public synchronized void refresh() throws IOException {
loadAll(requireActiveService());
journalAdmin.refresh();
}
private JournalServiceImpl requireActiveService() {
if (activeService == null) {
throw new IllegalStateException("Journal service manager has not been started");
}
return activeService;
}
private void loadAll(JournalServiceImpl service) throws IOException {
validateRoot();
for (Path artifactDirectory : listEntries(journalRoot)) {
if (!Files.isDirectory(artifactDirectory)) {
throw new IOException("Unexpected entry in Journal root; expected an artifact directory: "
+ artifactDirectory);
}
loadArtifactDirectory(service, artifactDirectory);
}
}
private void validateRoot() throws IOException {
if (!Files.exists(journalRoot)) {
throw new IOException("Configured Journal root does not exist: " + journalRoot);
}
if (!Files.isDirectory(journalRoot)) {
throw new IOException("Configured Journal root is not a directory: " + journalRoot);
}
if (!Files.isReadable(journalRoot)) {
throw new IOException("Configured Journal root is not readable: " + journalRoot);
}
}
private static void loadArtifactDirectory(
JournalServiceImpl service,
Path directory) throws IOException {
String directoryName = directory.getFileName().toString();
ArtifactCoordinate expectedCoordinate;
try {
expectedCoordinate = ArtifactCoordinate.parse(directoryName);
} catch (IllegalArgumentException exception) {
throw new IOException("Invalid artifact-coordinate directory name: " + directory, exception);
}
if (!Files.isReadable(directory)) {
throw new IOException("Artifact Journal directory is not readable: " + directory);
}
ArrayList<JournalFile> runtimeFiles = new ArrayList<>();
for (Path entry : listEntries(directory)) {
if (!Files.isRegularFile(entry)) {
throw new IOException("Unexpected entry in artifact Journal directory: " + entry);
}
String filename = entry.getFileName().toString();
if (filename.endsWith(EXAMPLE_SUFFIX)) {
continue;
}
if (!filename.endsWith(JOURNAL_SUFFIX)) {
throw new IOException("Unexpected file in artifact Journal directory: " + entry);
}
if (!Files.isReadable(entry)) {
throw new IOException("Journal file is not readable: " + entry);
}
String versionText = filename.substring(0, filename.length() - JOURNAL_SUFFIX.length());
JournalVersion expectedVersion;
try {
expectedVersion = JournalVersion.parse(versionText);
} catch (IllegalArgumentException exception) {
throw new IOException("Invalid Journal filename; expected <journal-version>.journal: " + entry,
exception);
}
runtimeFiles.add(new JournalFile(entry, expectedVersion));
}
runtimeFiles.sort(Comparator.comparing(JournalFile::journalVersion));
for (JournalFile runtimeFile : runtimeFiles) {
String text = Files.readString(runtimeFile.path, StandardCharsets.UTF_8);
String source = runtimeFile.path.toString();
Journal parsed = service.parse(text, source);
if (!parsed.metadata().coordinate().equals(expectedCoordinate)) {
throw new IOException("Journal metadata coordinate " + parsed.metadata().coordinate()
+ " does not match directory " + expectedCoordinate + ": " + runtimeFile.path);
}
if (!parsed.journalVersion().equals(runtimeFile.journalVersion)) {
throw new IOException("JOURNAL_VERSION " + parsed.journalVersion()
+ " does not match filename version " + runtimeFile.journalVersion + ": " + runtimeFile.path);
}
service.ingest(text, source);
}
}
private static List<Path> listEntries(Path directory) throws IOException {
try (Stream<Path> entries = Files.list(directory)) {
return entries.sorted(Comparator.comparing(path -> path.getFileName().toString())).toList();
} catch (SecurityException exception) {
throw new IOException("Cannot read Journal directory: " + directory, exception);
}
}
private record JournalFile(Path path, JournalVersion journalVersion) {
private JournalFile {
Objects.requireNonNull(path, "path");
Objects.requireNonNull(journalVersion, "journalVersion");
}
}
private final JournalServiceAdmin journalAdmin;
}