diff --git a/AGENTS.md b/AGENTS.md index 57ff965c3e..86bd50d6bd 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -43,4 +43,7 @@ - For serverless changes, ensure planner/invoker/worker settings remain consistent (input/intermediate/output storage schemes in `pixels-turbo/README.md`). ## Optional skills -- For guided Pixels deployment, install and use the `pixels-install` skill from `skills/pixels-install`. +- For guided Pixels deployment, install and use the `pixels-install` skill from `skills/pixels-install`: + - Cursor: `./skills/install.sh --skill pixels-install --tool cursor --scope project` + - Codex: `./skills/install.sh --skill pixels-install --tool codex --scope project` + - Claude: `./skills/install.sh --skill pixels-install --tool claude --scope project` diff --git a/pixels-common/src/main/resources/pixels.properties b/pixels-common/src/main/resources/pixels.properties index 3e86fb9680..74915a9882 100644 --- a/pixels-common/src/main/resources/pixels.properties +++ b/pixels-common/src/main/resources/pixels.properties @@ -279,8 +279,6 @@ retina.enable=false retina.tile.visibility.capacity=10240 # number of rows recorded in memTable, must be a multiple of 64 retina.buffer.memTable.size=10240 -# the scheme of the storage for retina buffer object storage, e.g., s3, minio -retina.buffer.object.storage.scheme=s3 # the folder path for retina buffer object storage retina.buffer.object.storage.folder=s3://object-storage # number of threads to flush shared storage diff --git a/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/Daemon.java b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/Daemon.java index bccc8798d4..fc8c9ffed6 100644 --- a/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/Daemon.java +++ b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/Daemon.java @@ -67,7 +67,7 @@ public void setup (String lockFilePath) if (myLock == null) { // this process has been started - this.clean(); + this.releaseLock(); log.info("another daemon process is holding lock on " + lockFilePath + ", exiting..."); this.running = false; System.exit(0); @@ -82,8 +82,8 @@ public void setup (String lockFilePath) * Issue #181: * We should not bind the SIGTERM handler. * If it is bind, the shutdown hooks will not be called. - * Daemon.shutdown() will be called in the shutdown hook in DaemonMain, - * instead of in the signal handler. + * Daemon shutdown is coordinated by the shutdown hook in DaemonMain, + * instead of by the signal handler. */ // bind handler for SIGTERM(15) signal. //this.shutdownHandler = new ShutdownHandler(this); @@ -91,45 +91,65 @@ public void setup (String lockFilePath) } catch (IOException e) { log.error("I/O exception when creating lock file channels.", e); - this.clean(); + this.releaseLock(); } } - public void clean () + /** + * Stop the daemon main loop without releasing the process lock. + */ + public void requestShutdown() { + this.running = false; + } + + /** + * Release the process lock after all managed servers have been stopped. + */ + public synchronized void releaseLock() + { + if (this.cleaned) + { + return; + } + log.info("releasing daemon lock and cleaning the file channel..."); try { - if (myLock != null) + if (this.myLock != null && this.myLock.isValid()) { - myLock.release(); + this.myLock.release(); } } catch (IOException e1) { - log.error("error when releasing daemon lock"); + log.error("error when releasing daemon lock", e1); } - try + if (this.myChannel != null) { - if (this.myChannel != null) + try { this.myChannel.truncate(0); + } + catch (IOException e) + { + log.error("error when truncating my own channel", e); + } + try + { this.myChannel.close(); } - } catch (IOException e) - { - log.error("error when closing my own channel.", e); + catch (IOException e) + { + log.error("error when closing my own channel", e); + } } this.cleaned = true; } public void shutdown() { - this.running = false; - while (!this.cleaned) - { - log.info("the daemon thread is stopped, cleaning the file channels..."); - this.clean(); - } + this.requestShutdown(); + this.releaseLock(); } public boolean isRunning() @@ -153,8 +173,15 @@ public void handle(Signal signal) { if (signal.getNumber() == 15) { - this.daemon.shutdown(); - this.executor.run(); + this.daemon.requestShutdown(); + try + { + this.executor.run(); + } + finally + { + this.daemon.releaseLock(); + } } } } diff --git a/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/DaemonMain.java b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/DaemonMain.java index c2ca3b9e34..baae9c1965 100644 --- a/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/DaemonMain.java +++ b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/DaemonMain.java @@ -6,10 +6,11 @@ import io.pixelsdb.pixels.daemon.cache.CacheWorker; import io.pixelsdb.pixels.daemon.index.IndexServer; import io.pixelsdb.pixels.daemon.node.NodeServer; +import io.pixelsdb.pixels.daemon.retina.RetinaReadyCheck; import io.pixelsdb.pixels.daemon.retina.RetinaServer; -import io.pixelsdb.pixels.daemon.exception.NoSuchServerException; import io.pixelsdb.pixels.daemon.heartbeat.HeartbeatCoordinator; import io.pixelsdb.pixels.daemon.heartbeat.HeartbeatWorker; +import io.pixelsdb.pixels.daemon.metadata.MetadataReadyCheck; import io.pixelsdb.pixels.daemon.metadata.MetadataServer; import io.pixelsdb.pixels.daemon.metrics.MetricsServer; import io.pixelsdb.pixels.daemon.scaling.ScalingMetricsServer; @@ -108,7 +109,7 @@ public static void main(String[] args) container.addServer("metadata", metadataServer); // start transaction server TransServer transServer = new TransServer(transServerPort); - container.addServer("transaction", transServer); + container.addServer("transaction", transServer, new RetinaReadyCheck()); // start query schedule server QueryScheduleServer queryScheduleServer = new QueryScheduleServer(queryScheduleServerPort); container.addServer("query_schedule", queryScheduleServer); @@ -174,7 +175,8 @@ else if(role.equals(NodeProto.NodeRole.RETINA)) { // start retina server on worker node RetinaServer retinaServer = new RetinaServer(retinaServerPort); - container.addServer("retina", retinaServer); + container.addServer("retina", retinaServer, + new MetadataReadyCheck()); if (indexServerEnabled) { @@ -202,47 +204,27 @@ else if(role.equals(NodeProto.NodeRole.RETINA)) // The shutdown hook ensures the servers are shutdown graceful // if this main daemon is terminated by SIGTERM(15) signal. ShutdownHookManager.Instance().registerShutdownHook(DaemonMain.class, false, () -> { - for (String serverName : container.getServerNames()) + // Stop the watchdog before shutting down servers, but retain the + // process lock until all server shutdown attempts have completed. + mainDaemon.requestShutdown(); + try { - // shutdown the server threads. - try - { - container.shutdownServer(serverName); - } catch (NoSuchServerException e) + container.shutdownAll(); + if (!container.awaitTermination(30, TimeUnit.SECONDS)) { - log.error("error when stopping server threads", e); + log.warn("not all servers stopped before the shutdown timeout"); } } - for (int i = 60; i > 0; --i) + catch (InterruptedException e) { - try - { - boolean done = true; - for (String serverName : container.getServerNames()) - { - if (container.checkServer(serverName, 0)) - { - done = false; - break; - } - } - if (done) - { - break; - } - TimeUnit.SECONDS.sleep(1); - } - catch (Throwable e) - { - log.error("error when waiting server threads shutdown", e); - } + Thread.currentThread().interrupt(); + log.warn("interrupted while waiting for servers to stop", e); } - /** - * Issue #181: - * Shutdown the daemon thread here instead of using the SIGTERM handler. - */ - mainDaemon.shutdown(); - log.info("all the servers are shutdown, bye..."); + finally + { + mainDaemon.releaseLock(); + } + log.info("daemon shutdown completed, bye..."); }); // continue the main thread, start and check the server threads. @@ -253,10 +235,7 @@ else if(role.equals(NodeProto.NodeRole.RETINA)) { for (String serverName : container.getServerNames()) { - if (!container.checkServer(serverName)) - { - container.startServer(serverName); - } + container.startServer(serverName); } TimeUnit.SECONDS.sleep(1); } diff --git a/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/ServerContainer.java b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/ServerContainer.java index e0d11be1a9..463e7b8eaf 100644 --- a/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/ServerContainer.java +++ b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/ServerContainer.java @@ -25,6 +25,7 @@ import org.apache.logging.log4j.Logger; import java.util.ArrayList; +import java.util.Arrays; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -37,95 +38,316 @@ public class ServerContainer { private static Logger log = LogManager.getLogger(ServerContainer.class); - private Map serverMap = null; + private static final class ServerHandle + { + private final Server server; + private final List startupChecks; + private Thread thread; + private boolean shutdownInvoked; + + private ServerHandle(Server server, List startupChecks) + { + this.server = server; + this.startupChecks = startupChecks; + } + } + private final Map serverHandles; + /** + * Once shutdown begins, no server thread can be (re)started in this container, + * otherwise the daemon main loop would resurrect the servers that are being shutdown. + */ + private boolean shuttingDown = false; public ServerContainer () { - this.serverMap = new HashMap<>(); + this.serverHandles = new HashMap<>(); } - public void addServer (String name, Server server) + public synchronized void addServer ( + String name, Server server, StartupCheck... startupChecks) { - Thread thread = new Thread(server); - thread.start(); - this.serverMap.put(name, server); + if (this.shuttingDown) + { + throw new IllegalStateException("server container is shutting down"); + } + if (this.serverHandles.containsKey(name)) + { + throw new IllegalArgumentException("server is already registered: " + name); + } + this.serverHandles.put(name, new ServerHandle(server, Arrays.asList(startupChecks))); + startServerThread(name); } - public List getServerNames() + public synchronized List getServerNames() { - return new ArrayList<>(this.serverMap.keySet()); + return new ArrayList<>(this.serverHandles.keySet()); } /** - * retry 3 times by default. sleep one second after each retry. - * @param name - * @return - * @throws NoSuchServerException + * Ensure that a server has one thread responsible for its lifecycle. + * A server may still be waiting for startup checks while its + * {@link Server#isRunning()} method returns false. */ - public boolean checkServer(String name) throws NoSuchServerException + public synchronized void startServer(String name) throws NoSuchServerException { - return this.checkServer(name, 3); + ServerHandle handle = this.serverHandles.get(name); + if (handle == null) + { + throw new NoSuchServerException(); + } + if (this.shuttingDown) + { + log.debug("Server container is shutting down, skip starting {}", name); + return; + } + Thread serverThread = handle.thread; + if ((serverThread != null && serverThread.isAlive()) + || handle.server.isRunning()) + { + log.debug("Server {} is already starting or running, skip duplicate start", name); + return; + } + startServerThread(name); } /** - * - * @param name - * @param retry times to retry, sleep one second after each retry. - * @return true if server is running. - * @throws NoSuchServerException + * Check whether the thread responsible for a server's lifecycle is alive. */ - public boolean checkServer(String name, int retry) throws NoSuchServerException + public synchronized boolean checkServer(String name) + throws NoSuchServerException { - Server server = this.serverMap.get(name); - if (server == null) + ServerHandle handle = this.serverHandles.get(name); + if (handle == null) { throw new NoSuchServerException(); } - boolean serverIsRunning = false; - try + Thread serverThread = handle.thread; + return serverThread != null && serverThread.isAlive(); + } + + /** + * Atomically prevent new server starts and initiate shutdown of every registered server. + */ + public void shutdownAll() + { + Map handlesToShutdown; + synchronized (this) { - if (!server.isRunning()) + if (this.shuttingDown) { - for (int i = 0; i < retry; ++i) + return; + } + this.shuttingDown = true; + handlesToShutdown = new HashMap<>(this.serverHandles); + } + + for (Map.Entry entry : handlesToShutdown.entrySet()) + { + shutdownServer(entry.getKey(), entry.getValue()); + } + } + + /** + * Wait until all lifecycle threads have exited and all servers report stopped. + * + * @return true if every server stopped before the timeout, false otherwise + */ + public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException + { + if (timeout < 0) + { + throw new IllegalArgumentException("timeout is negative"); + } + long timeoutNanos = unit.toNanos(timeout); + long startNanos = System.nanoTime(); + + while (true) + { + Map handles; + boolean shutdownStarted; + synchronized (this) + { + handles = new HashMap<>(this.serverHandles); + shutdownStarted = this.shuttingDown; + } + + List unterminatedServers = new ArrayList<>(); + for (Map.Entry entry : handles.entrySet()) + { + String name = entry.getKey(); + ServerHandle handle = entry.getValue(); + boolean running = isServerRunning(name, handle); + if (shutdownStarted && running) { - // try 3 times - TimeUnit.SECONDS.sleep(1); - if (server.isRunning()) - { - serverIsRunning = true; - break; - } + invokeShutdown(name, handle); + } + Thread serverThread = getServerThread(handle); + if ((serverThread != null && serverThread.isAlive()) || running) + { + unterminatedServers.add(name); } } - else + + if (unterminatedServers.isEmpty()) + { + return true; + } + + long remainingNanos = timeoutNanos - (System.nanoTime() - startNanos); + if (remainingNanos <= 0) + { + log.warn("Timed out waiting for servers to stop: {}", unterminatedServers); + return false; + } + + synchronized (this) { - serverIsRunning = true; + TimeUnit.NANOSECONDS.timedWait( + this, Math.min(remainingNanos, TimeUnit.MILLISECONDS.toNanos(100))); } - } catch (InterruptedException e) + } + } + + private void shutdownServer(String name, ServerHandle handle) + { + if (isServerRunning(name, handle)) { - log.error( - "interrupted while checking server.", e); + invokeShutdown(name, handle); + } + else + { + interruptServerThread(handle); } - return serverIsRunning; } - public void startServer(String name) throws NoSuchServerException + private void invokeShutdown(String name, ServerHandle handle) { - this.shutdownServer(name); - Thread serverThread = new Thread(this.serverMap.get(name)); - serverThread.start(); + synchronized (this) + { + if (handle.shutdownInvoked) + { + return; + } + handle.shutdownInvoked = true; + } + try + { + handle.server.shutdown(); + } + catch (Throwable e) + { + log.error("Failed to shutdown server {}", name, e); + } + finally + { + interruptServerThread(handle); + } } - public void shutdownServer(String name) throws NoSuchServerException + private boolean isServerRunning(String name, ServerHandle handle) { - Server server = this.serverMap.get(name); - if (server == null) + try { - throw new NoSuchServerException(); + return handle.server.isRunning(); + } + catch (Throwable e) + { + log.error("Failed to check whether server {} is running", name, e); + return true; + } + } + + private synchronized Thread getServerThread(ServerHandle handle) + { + return handle.thread; + } + + private void interruptServerThread(ServerHandle handle) + { + Thread serverThread = getServerThread(handle); + if (serverThread != null && serverThread.isAlive() + && serverThread != Thread.currentThread()) + { + serverThread.interrupt(); + } + } + + private void startServerThread(String name) + { + ServerHandle handle = this.serverHandles.get(name); + if (handle == null) + { + throw new IllegalStateException("server is not registered: " + name); } - if (checkServer(name, 0) == true) + Thread existingThread = handle.thread; + if (existingThread != null && existingThread.isAlive()) { - server.shutdown(); + return; } + + handle.thread = new Thread(() -> + { + try + { + long startupDeadline = System.nanoTime() + 60_000_000_000L; + for (StartupCheck startupCheck : handle.startupChecks) + { + long remainingNanos = startupDeadline - System.nanoTime(); + if (remainingNanos <= 0) + { + throw new IllegalStateException( + "Timed out waiting for startup checks of " + name + + " after 60 seconds"); + } + log.debug("Server {} is waiting for {}", name, startupCheck.getDescription()); + startupCheck.awaitReady(startupDeadline); + if (System.nanoTime() >= startupDeadline) + { + throw new IllegalStateException( + "Timed out waiting for startup checks of " + name + + " after 60 seconds"); + } + } + synchronized (ServerContainer.this) + { + if (ServerContainer.this.shuttingDown + || Thread.currentThread().isInterrupted()) + { + return; + } + } + handle.server.run(); + } + catch (InterruptedException e) + { + Thread.currentThread().interrupt(); + log.info("Server {} startup was interrupted", name); + } + catch (Throwable e) + { + log.error("Server {} failed during startup or execution", name, e); + } + finally + { + boolean shutdownStarted; + synchronized (ServerContainer.this) + { + shutdownStarted = ServerContainer.this.shuttingDown; + } + if (shutdownStarted) + { + invokeShutdown(name, handle); + } + synchronized (ServerContainer.this) + { + if (handle.thread == Thread.currentThread()) + { + handle.thread = null; + } + ServerContainer.this.notifyAll(); + } + } + }, name); + handle.thread.start(); } } diff --git a/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/StartupCheck.java b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/StartupCheck.java new file mode 100644 index 0000000000..44dd465c30 --- /dev/null +++ b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/StartupCheck.java @@ -0,0 +1,41 @@ +/* + * Copyright 2026 PixelsDB. + * + * This file is part of Pixels. + * + * Pixels is free software: you can redistribute it and/or modify + * it under the terms of the Affero GNU General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * Pixels is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Affero GNU General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public + * License along with Pixels. If not, see + * . + */ +package io.pixelsdb.pixels.daemon; + +/** + * A check that must complete before a server starts serving requests. + * + *

The wait is interruptible so that a daemon shutdown can cancel a server + * that is still waiting for one of its startup checks.

+ * + * @author PixelsDB + */ +public interface StartupCheck +{ + String getDescription(); + + /** + * Wait until this startup check is satisfied or the supplied deadline expires. + * + * @param deadlineNanos an absolute deadline from {@link System#nanoTime()} + * @throws InterruptedException if the wait is interrupted + */ + void awaitReady(long deadlineNanos) throws InterruptedException; +} diff --git a/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/cache/CacheCoordinator.java b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/cache/CacheCoordinator.java index 4b311dd741..9a2640a538 100644 --- a/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/cache/CacheCoordinator.java +++ b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/cache/CacheCoordinator.java @@ -72,7 +72,7 @@ public class CacheCoordinator implements Server private Storage storage = null; private boolean initializeSuccess = false; private CountDownLatch runningLatch; - private boolean running = false; + private volatile boolean running = false; public CacheCoordinator() { @@ -154,6 +154,7 @@ public void run() return; } + CountDownLatch lifecycleLatch = new CountDownLatch(1); synchronized (this) { if (this.running) @@ -162,12 +163,12 @@ public void run() } else { + this.runningLatch = lifecycleLatch; this.running = true; } } logger.info("Starting cache coordinator"); - runningLatch = new CountDownLatch(1); // watch layout version change, and update the cache plan and the local cache version Watch.Watcher watcher = EtcdUtil.Instance().getClient().getWatchClient().watch( ByteSequence.from(Constants.LAYOUT_VERSION_LITERAL, StandardCharsets.UTF_8), @@ -209,7 +210,7 @@ public void run() { // Wait for this coordinator to be shutdown logger.info("Cache coordinator is running"); - runningLatch.await(); + lifecycleLatch.await(); } catch (InterruptedException e) { @@ -217,6 +218,7 @@ public void run() } finally { + this.running = false; if (watcher != null) { watcher.close(); diff --git a/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/heartbeat/HeartbeatWorker.java b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/heartbeat/HeartbeatWorker.java index ac358c05f1..275e3d840c 100644 --- a/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/heartbeat/HeartbeatWorker.java +++ b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/heartbeat/HeartbeatWorker.java @@ -66,7 +66,8 @@ public static void setCurrentStatus(NodeStatus status) { throw new IllegalArgumentException("status is null"); } - currentStatus.set(status.StatusCode); + currentStatus.updateAndGet(current -> + current == NodeStatus.EXIT.StatusCode ? current : status.StatusCode); } /** diff --git a/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/index/IndexServer.java b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/index/IndexServer.java index ce412ef453..bb91680dd0 100644 --- a/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/index/IndexServer.java +++ b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/index/IndexServer.java @@ -36,7 +36,7 @@ public class IndexServer implements Server { private static final Logger log = LogManager.getLogger(IndexServer.class); - private boolean running = false; + private volatile boolean running = false; private final io.grpc.Server rpcServer; public IndexServer(int port) diff --git a/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/metadata/MetadataReadyCheck.java b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/metadata/MetadataReadyCheck.java new file mode 100644 index 0000000000..d3cd04fb2c --- /dev/null +++ b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/metadata/MetadataReadyCheck.java @@ -0,0 +1,114 @@ +/* + * Copyright 2026 PixelsDB. + * + * This file is part of Pixels. + * + * Pixels is free software: you can redistribute it and/or modify + * it under the terms of the Affero GNU General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * Pixels is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Affero GNU General Public License for more details. + * + * You should have received a copy of the Affero GNU General Public + * License along with Pixels. If not, see + * . + */ +package io.pixelsdb.pixels.daemon.metadata; + +import io.grpc.ManagedChannel; +import io.grpc.ManagedChannelBuilder; +import io.grpc.StatusRuntimeException; +import io.grpc.health.v1.HealthCheckRequest; +import io.grpc.health.v1.HealthCheckResponse; +import io.grpc.health.v1.HealthGrpc; +import io.pixelsdb.pixels.common.utils.ConfigFactory; +import io.pixelsdb.pixels.daemon.StartupCheck; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.util.concurrent.TimeUnit; + +/** + * Waits until the Metadata gRPC health service reports SERVING. + * + * @author PixelsDB + */ +public class MetadataReadyCheck implements StartupCheck +{ + private static final Logger log = LogManager.getLogger(MetadataReadyCheck.class); + private static final long HEALTH_CHECK_TIMEOUT_MS = 1_000L; + private static final long RETRY_INTERVAL_MS = 1_000L; + + @Override + public String getDescription() + { + return "Metadata gRPC health service to report SERVING"; + } + + @Override + public void awaitReady(long deadlineNanos) throws InterruptedException + { + ConfigFactory config = ConfigFactory.Instance(); + String host = config.getProperty("metadata.server.host"); + int port = Integer.parseInt(config.getProperty("metadata.server.port")); + + ManagedChannel channel = ManagedChannelBuilder.forAddress(host, port) + .usePlaintext() + .build(); + try + { + HealthGrpc.HealthBlockingStub stub = HealthGrpc.newBlockingStub(channel); + log.info("Waiting for Metadata server {}:{} to report SERVING", host, port); + while (true) + { + long remainingNanos = deadlineNanos - System.nanoTime(); + if (remainingNanos <= 0) + { + throw new IllegalStateException( + "Timed out waiting for Metadata server " + host + ":" + port + + " to report SERVING"); + } + try + { + HealthCheckResponse response = stub + .withDeadlineAfter( + Math.min( + TimeUnit.MILLISECONDS.toNanos(HEALTH_CHECK_TIMEOUT_MS), + remainingNanos), + TimeUnit.NANOSECONDS) + .check(HealthCheckRequest.newBuilder().setService("metadata").build()); + if (response.getStatus() == HealthCheckResponse.ServingStatus.SERVING) + { + log.info("Metadata server {}:{} is ready", host, port); + return; + } + } + catch (StatusRuntimeException e) + { + if (Thread.currentThread().isInterrupted()) + { + throw new InterruptedException("Interrupted while checking Metadata readiness"); + } + log.debug("Metadata health check failed: {}", e.getStatus()); + } + remainingNanos = deadlineNanos - System.nanoTime(); + if (remainingNanos <= 0) + { + throw new IllegalStateException( + "Timed out waiting for Metadata server " + host + ":" + port + + " to report SERVING"); + } + TimeUnit.NANOSECONDS.sleep(Math.min( + TimeUnit.MILLISECONDS.toNanos(RETRY_INTERVAL_MS), remainingNanos)); + } + } + finally + { + channel.shutdownNow(); + } + } +} diff --git a/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/metadata/MetadataServer.java b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/metadata/MetadataServer.java index d9a2a4d043..2ccf04a9ff 100644 --- a/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/metadata/MetadataServer.java +++ b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/metadata/MetadataServer.java @@ -40,7 +40,7 @@ public class MetadataServer implements Server { private static final Logger log = LogManager.getLogger(MetadataServer.class); - private boolean running = false; + private volatile boolean running = false; private final io.grpc.Server rpcServer; private final HealthStatusManager health = new HealthStatusManager(); diff --git a/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/metrics/MetricsServer.java b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/metrics/MetricsServer.java index ad75c72220..b9172c2bf4 100644 --- a/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/metrics/MetricsServer.java +++ b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/metrics/MetricsServer.java @@ -42,7 +42,7 @@ public class MetricsServer implements Server { private static Logger log = LogManager.getLogger(MetricsServer.class); - private boolean running = false; + private volatile boolean running = false; @Override public boolean isRunning() diff --git a/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/node/NodeServer.java b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/node/NodeServer.java index f889abb5d0..6bcb22add5 100644 --- a/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/node/NodeServer.java +++ b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/node/NodeServer.java @@ -32,7 +32,7 @@ public class NodeServer implements Server { private static final Logger log = LogManager.getLogger(NodeServer.class); - private boolean running = false; + private volatile boolean running = false; private final io.grpc.Server rpcServer; public NodeServer(int port) diff --git a/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/retina/RetinaReadyCheck.java b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/retina/RetinaReadyCheck.java new file mode 100644 index 0000000000..6413672399 --- /dev/null +++ b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/retina/RetinaReadyCheck.java @@ -0,0 +1,197 @@ +/* + * Copyright 2026 PixelsDB. + * + * This file is part of Pixels. + * + * Pixels is free software: you can redistribute it and/or modify + * it under the terms of the Affero GNU General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * Pixels is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Affero GNU General Public License for more details. + * + * You should have received a copy of the Affero GNU General Public + * License along with Pixels. If not, see + * . + */ +package io.pixelsdb.pixels.daemon.retina; + +import io.etcd.jetcd.ByteSequence; +import io.etcd.jetcd.KeyValue; +import io.etcd.jetcd.options.GetOption; +import io.pixelsdb.pixels.common.utils.ConfigFactory; +import io.pixelsdb.pixels.common.utils.Constants; +import io.pixelsdb.pixels.common.utils.EtcdUtil; +import io.pixelsdb.pixels.daemon.StartupCheck; +import io.pixelsdb.pixels.daemon.heartbeat.NodeStatus; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.HashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +/** + * Waits until every configured Retina node has a leased READY heartbeat. + * + * @author PixelsDB + */ +public class RetinaReadyCheck implements StartupCheck +{ + private static final Logger log = LogManager.getLogger(RetinaReadyCheck.class); + private static final long ETCD_READ_TIMEOUT_MS = 1_000L; + private static final long RETRY_INTERVAL_MS = 1_000L; + + @Override + public String getDescription() + { + return "all configured Retina nodes to report READY"; + } + + @Override + public void awaitReady(long deadlineNanos) throws InterruptedException + { + ConfigFactory config = ConfigFactory.Instance(); + if (!Boolean.parseBoolean(config.getProperty("retina.enable"))) + { + return; + } + + Set expected = loadExpectedNodes(config); + if (expected.isEmpty()) + { + throw new IllegalStateException( + "retina.enable=true but $PIXELS_HOME/etc/retina has no nodes"); + } + + String lastReason = null; + log.info("Waiting for {} Retina node(s) to report READY", expected.size()); + + while (true) + { + long remainingNanos = deadlineNanos - System.nanoTime(); + if (remainingNanos <= 0) + { + throw new IllegalStateException("Timed out waiting for Retina readiness"); + } + String reason = getReadinessReason(expected, remainingNanos); + if (reason == null) + { + log.info("All Retina nodes are READY"); + return; + } + if (!reason.equals(lastReason)) + { + log.info("Retina readiness is pending: {}", reason); + lastReason = reason; + } + remainingNanos = deadlineNanos - System.nanoTime(); + if (remainingNanos <= 0) + { + throw new IllegalStateException( + "Timed out waiting for Retina readiness; last reason: " + reason); + } + TimeUnit.NANOSECONDS.sleep(Math.min( + TimeUnit.MILLISECONDS.toNanos(RETRY_INTERVAL_MS), remainingNanos)); + } + } + + private Set loadExpectedNodes(ConfigFactory config) + { + String pixelsHome = config.getProperty("pixels.home"); + if (pixelsHome == null || pixelsHome.isEmpty()) + { + throw new IllegalStateException("pixels.home is not configured"); + } + Path retinaFile = Paths.get(pixelsHome, "etc", "retina"); + if (!Files.isRegularFile(retinaFile)) + { + throw new IllegalStateException(retinaFile + " is missing"); + } + + Set expected = new LinkedHashSet<>(); + try + { + for (String raw : Files.readAllLines(retinaFile, StandardCharsets.UTF_8)) + { + String line = raw.trim(); + if (line.isEmpty() || line.startsWith("#")) + { + continue; + } + expected.add(line.split("\\s+", 2)[0]); + } + } + catch (Exception e) + { + throw new IllegalStateException( + "Failed to load expected Retina nodes from " + retinaFile, e); + } + return expected; + } + + private String getReadinessReason(Set expected, long timeoutNanos) + throws InterruptedException + { + String prefix = Constants.HEARTBEAT_RETINA_LITERAL; + Map observed; + try + { + ByteSequence prefixBytes = ByteSequence.from(prefix, StandardCharsets.UTF_8); + GetOption getOption = GetOption.builder().isPrefix(true).build(); + List all = EtcdUtil.Instance().getClient().getKVClient() + .get(prefixBytes, getOption) + .get(Math.min( + TimeUnit.MILLISECONDS.toNanos(ETCD_READ_TIMEOUT_MS), + timeoutNanos), + TimeUnit.NANOSECONDS) + .getKvs(); + observed = new HashMap<>(all.size() * 2); + for (KeyValue kv : all) + { + String key = kv.getKey().toString(StandardCharsets.UTF_8); + if (key.length() > prefix.length()) + { + observed.put(key.substring(prefix.length()), kv); + } + } + } + catch (InterruptedException e) + { + throw e; + } + catch (Exception e) + { + return "Etcd heartbeat read failed: " + e.getMessage(); + } + + for (String host : expected) + { + KeyValue kv = observed.get(host); + if (kv == null) + { + return "Retina node " + host + " has no heartbeat status"; + } + if (kv.getLease() <= 0) + { + return "Retina node " + host + " has heartbeat status without lease"; + } + String status = kv.getValue().toString(StandardCharsets.UTF_8).trim(); + if (!String.valueOf(NodeStatus.READY.StatusCode).equals(status)) + { + return "Retina node " + host + " heartbeat status is " + status; + } + } + return null; + } +} diff --git a/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/retina/RetinaServer.java b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/retina/RetinaServer.java index ab1697d4da..5ce26563bc 100644 --- a/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/retina/RetinaServer.java +++ b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/retina/RetinaServer.java @@ -19,15 +19,10 @@ */ package io.pixelsdb.pixels.daemon.retina; -import io.grpc.ManagedChannel; -import io.grpc.ManagedChannelBuilder; import io.grpc.ServerBuilder; -import io.grpc.StatusRuntimeException; -import io.grpc.health.v1.HealthCheckRequest; -import io.grpc.health.v1.HealthCheckResponse; -import io.grpc.health.v1.HealthGrpc; import io.pixelsdb.pixels.common.server.Server; -import io.pixelsdb.pixels.common.utils.ConfigFactory; +import io.pixelsdb.pixels.daemon.heartbeat.HeartbeatWorker; +import io.pixelsdb.pixels.daemon.heartbeat.NodeStatus; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; @@ -44,16 +39,14 @@ public class RetinaServer implements Server { private static final Logger log = LogManager.getLogger(RetinaServer.class); - private boolean running = false; - private final io.grpc.Server rpcServer; + private volatile boolean running = false; + private final int port; + private volatile io.grpc.Server rpcServer; public RetinaServer(int port) { checkArgument(port > 0 && port <= 65535, "illegal rpc port"); - // Issue #935: ensure metadata server has been already started - waitForMetadataServer(); - this.rpcServer = ServerBuilder.forPort(port) - .addService(new RetinaServerImpl()).build(); + this.port = port; } @Override @@ -66,12 +59,16 @@ public boolean isRunning() public void shutdown() { this.running = false; - try - { - this.rpcServer.shutdown().awaitTermination(5, TimeUnit.SECONDS); - } catch (InterruptedException e) + io.grpc.Server server = this.rpcServer; + if (server != null) { - log.error("Interrupted when shutdown rpc server", e); + try + { + server.shutdown().awaitTermination(5, TimeUnit.SECONDS); + } catch (InterruptedException e) + { + log.error("Interrupted when shutdown rpc server", e); + } } } @@ -80,9 +77,16 @@ public void run() { try { - this.rpcServer.start(); + HeartbeatWorker.setCurrentStatus(NodeStatus.INIT); + RetinaServerImpl service = new RetinaServerImpl(); + service.setReadyListener(() -> publishReady(service)); + io.grpc.Server server = ServerBuilder.forPort(port) + .addService(service).build(); + this.rpcServer = server; + server.start(); this.running = true; - this.rpcServer.awaitTermination(); + publishReady(service); + server.awaitTermination(); } catch (IOException e) { log.error("I/O error when running", e); @@ -95,49 +99,12 @@ public void run() } } - private void waitForMetadataServer() + private void publishReady(RetinaServerImpl service) { - ConfigFactory config = ConfigFactory.Instance(); - String metadataHost = config.getProperty("metadata.server.host"); - int metadataPort = Integer.parseInt(config.getProperty("metadata.server.port")); - ManagedChannel channel = ManagedChannelBuilder.forAddress(metadataHost, metadataPort) - .usePlaintext().build(); - - HealthGrpc.HealthBlockingStub stub = HealthGrpc.newBlockingStub(channel); - int retry = 0; - int maxRetry = 30; - - while (retry < maxRetry) - { - try - { - HealthCheckResponse response = stub.check(HealthCheckRequest.newBuilder().setService("metadata").build()); - if (response.getStatus() == HealthCheckResponse.ServingStatus.SERVING) - { - log.info("Metadata server is ready"); - channel.shutdown(); - return; - } - } catch (StatusRuntimeException e) - { - log.info("Metadata health check failed, sleep one second and retry"); - } - - retry++; - try - { - Thread.sleep(1000); - } catch (InterruptedException e) - { - log.error("Failed to sleep for retry to check metadata server status", e); - Thread.currentThread().interrupt(); - break; - } - } - channel.shutdown(); - if (retry >= maxRetry) + if (this.running && service.isReady()) { - log.error("Timeout waiting for metadata server to be ready"); + HeartbeatWorker.setCurrentStatus(NodeStatus.READY); + log.info("Retina service is ready"); } } } diff --git a/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/retina/RetinaServerImpl.java b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/retina/RetinaServerImpl.java index 1bc27d61e3..66fbbfa3cc 100644 --- a/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/retina/RetinaServerImpl.java +++ b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/retina/RetinaServerImpl.java @@ -36,8 +36,6 @@ import io.pixelsdb.pixels.common.metadata.domain.*; import io.pixelsdb.pixels.common.utils.ConfigFactory; import io.pixelsdb.pixels.common.utils.IndexUtils; -import io.pixelsdb.pixels.daemon.heartbeat.HeartbeatWorker; -import io.pixelsdb.pixels.daemon.heartbeat.NodeStatus; import io.pixelsdb.pixels.index.IndexProto; import io.pixelsdb.pixels.retina.RGVisibility; import io.pixelsdb.pixels.retina.RecoveryCheckpoint; @@ -75,6 +73,7 @@ public class RetinaServerImpl extends RetinaWorkerServiceGrpc.RetinaWorkerServic private final RetinaResourceManager retinaResourceManager; private final Striped updateLocks = Striped.lock(1024); private volatile RetinaStatus status; + private volatile Runnable readyListener; private IndexOption[] indexOptionPool; /** @@ -110,7 +109,7 @@ public RetinaServerImpl() publishStartupLifecycle(recoveryContext, recoveryResult); startRetinaMetricsLogThread(); boolean ready = this.status.getState() == RetinaState.READY; - logger.info(ready ? "Retina service is ready" : "Retina service is recovering"); + logger.info(ready ? "Retina initialization is ready" : "Retina service is recovering"); } catch (Exception e) { @@ -122,12 +121,31 @@ public RetinaServerImpl() .setState(RetinaState.FAILED) .build(); this.retinaResourceManager.setRecovering(false); - HeartbeatWorker.setCurrentStatus(NodeStatus.INIT); logger.error("Error while initializing RetinaServerImpl", e); throw new IllegalStateException("Failed to initialize RetinaServerImpl", e); } } + public void setReadyListener(Runnable readyListener) + { + this.readyListener = readyListener; + } + + boolean isReady() + { + RetinaStatus current = this.status; + return current != null && current.getState() == RetinaState.READY; + } + + private void notifyReady() + { + Runnable listener = this.readyListener; + if (listener != null && isReady()) + { + listener.run(); + } + } + private RecoveryContext prepareRecoveryContext() throws RetinaException { String recoveryEpoch = UUID.randomUUID().toString(); @@ -357,7 +375,6 @@ private void initializeRecoveredResources() throws Exception private void publishStartupLifecycle(RecoveryContext context, RecoveryResult result) throws RetinaException { this.retinaResourceManager.setRecovering(true); - HeartbeatWorker.setCurrentStatus(NodeStatus.INIT); this.status = RetinaStatus.newBuilder() .setState(RetinaState.RECOVERING) .setRecoveryEpoch(context.recoveryEpoch) @@ -371,7 +388,9 @@ private void publishStartupLifecycle(RecoveryContext context, RecoveryResult res .setState(RetinaState.READY) .build(); this.retinaResourceManager.setRecovering(false); - HeartbeatWorker.setCurrentStatus(NodeStatus.READY); + // Do not notify the ready listener here. RetinaServerImpl finishes + // initialization before RetinaServer starts the gRPC server, so + // RetinaServer publishes READY only after the RPC server is available. } } @@ -1065,7 +1084,7 @@ public void markReady(RetinaProto.MarkReadyRequest request, .setState(RetinaState.READY) .build(); this.retinaResourceManager.setRecovering(false); - HeartbeatWorker.setCurrentStatus(NodeStatus.READY); + notifyReady(); } catch (RetinaException e) { diff --git a/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/scaling/ScalingMetricsServer.java b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/scaling/ScalingMetricsServer.java index 8c32305a6c..b37864ec69 100644 --- a/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/scaling/ScalingMetricsServer.java +++ b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/scaling/ScalingMetricsServer.java @@ -32,7 +32,7 @@ public class ScalingMetricsServer implements Server { private static final Logger log = LogManager.getLogger(ScalingMetricsServer.class); - private boolean running = false; + private volatile boolean running = false; private final io.grpc.Server rpcServer; private final int port; diff --git a/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/transaction/TransServer.java b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/transaction/TransServer.java index 05dd64192a..e51e33f9b8 100644 --- a/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/transaction/TransServer.java +++ b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/transaction/TransServer.java @@ -20,27 +20,13 @@ package io.pixelsdb.pixels.daemon.transaction; import java.io.IOException; -import java.nio.charset.StandardCharsets; -import java.nio.file.Files; -import java.nio.file.Path; -import java.nio.file.Paths; -import java.util.HashMap; -import java.util.LinkedHashSet; -import java.util.List; -import java.util.Map; -import java.util.Set; import java.util.concurrent.TimeUnit; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; -import io.etcd.jetcd.KeyValue; import io.grpc.ServerBuilder; import io.pixelsdb.pixels.common.server.Server; -import io.pixelsdb.pixels.common.utils.ConfigFactory; -import io.pixelsdb.pixels.common.utils.Constants; -import io.pixelsdb.pixels.common.utils.EtcdUtil; -import io.pixelsdb.pixels.daemon.heartbeat.NodeStatus; /** * @author hank @@ -50,14 +36,7 @@ public class TransServer implements Server { private static final Logger log = LogManager.getLogger(TransServer.class); - /** - * Default time to wait for all expected Retina nodes to reach READY before giving up - * and aborting the trans server boot. Overridable by {@code trans.server.retina.readiness.timeout.ms}. - */ - private static final long DEFAULT_RETINA_READINESS_TIMEOUT_MS = 10 * 60 * 1000L; - private static final long RETINA_READINESS_POLL_INTERVAL_MS = 1_000L; - - private boolean running = false; + private volatile boolean running = false; private final io.grpc.Server rpcServer; public TransServer(int port) @@ -91,7 +70,6 @@ public void run() { try { - awaitRetinaReady(); this.rpcServer.start(); this.running = true; this.rpcServer.awaitTermination(); @@ -106,126 +84,4 @@ public void run() this.shutdown(); } } - - /** - * Boot-time gate. When {@code retina.enable=true}, blocks until every node listed in - * {@code $PIXELS_HOME/etc/retina} reports {@code NodeStatus.READY} via heartbeat. When - * {@code retina.enable=false}, returns immediately. On timeout, throws so that - * {@link #run()} aborts and the supervisor can restart the process. - * - *

This is intentionally a one-shot check executed before the gRPC server starts. - * Once the trans server is serving, it does not re-check Retina lifecycle state. - */ - private void awaitRetinaReady() - { - ConfigFactory config = ConfigFactory.Instance(); - if (!Boolean.parseBoolean(config.getProperty("retina.enable"))) - { - return; - } - - // Load expected Retina nodes from $PIXELS_HOME/etc/retina. - Path retinaFile = Paths.get(config.getProperty("pixels.home"), "etc", "retina"); - if (!Files.isRegularFile(retinaFile)) - { - throw new IllegalStateException(retinaFile + " is missing"); - } - Set expected = new LinkedHashSet<>(); - try - { - for (String raw : Files.readAllLines(retinaFile, StandardCharsets.UTF_8)) - { - String line = raw.trim(); - if (line.isEmpty() || line.startsWith("#")) - { - continue; - } - String host = line.split("\\s+", 2)[0]; - expected.add(host); - } - } catch (IOException e) - { - throw new IllegalStateException("Failed to load expected Retina nodes from " - + "$PIXELS_HOME/etc/retina", e); - } - if (expected.isEmpty()) - { - throw new IllegalStateException( - "retina.enable=true but $PIXELS_HOME/etc/retina has no nodes"); - } - - long deadline = System.currentTimeMillis() + DEFAULT_RETINA_READINESS_TIMEOUT_MS; - EtcdUtil etcd = EtcdUtil.Instance(); - String prefix = Constants.HEARTBEAT_RETINA_LITERAL; - int prefixLen = prefix.length(); - log.info("Waiting for {} Retina node(s) to report READY (timeout {} ms)", - expected.size(), DEFAULT_RETINA_READINESS_TIMEOUT_MS); - while (true) - { - String reason = null; - // Poll all Retina heartbeat keys once and check whether every expected node is READY. - Map observed; - try - { - List all = etcd.getKeyValuesByPrefix(prefix); - observed = new HashMap<>(all.size() * 2); - for (KeyValue kv : all) - { - String key = kv.getKey().toString(StandardCharsets.UTF_8); - if (key.length() > prefixLen) - { - observed.put(key.substring(prefixLen), kv); - } - } - } catch (RuntimeException e) - { - observed = null; - reason = "etcd heartbeat read failed: " + e.getMessage(); - } - if (reason == null) - { - for (String host : expected) - { - KeyValue kv = observed.get(host); - if (kv == null) - { - reason = "Retina node " + host + " has no heartbeat status"; - break; - } - if (kv.getLease() <= 0) - { - reason = "Retina node " + host + " has heartbeat status without lease"; - break; - } - String status = kv.getValue().toString(StandardCharsets.UTF_8).trim(); - if (!String.valueOf(NodeStatus.READY.StatusCode).equals(status)) - { - reason = "Retina node " + host + " heartbeat status is " + status; - break; - } - } - } - if (reason == null) - { - log.info("All Retina nodes are READY, starting trans server"); - return; - } - if (System.currentTimeMillis() >= deadline) - { - throw new IllegalStateException( - "Timed out waiting for Retina readiness after " - + DEFAULT_RETINA_READINESS_TIMEOUT_MS - + " ms; last reason: " + reason); - } - try - { - Thread.sleep(RETINA_READINESS_POLL_INTERVAL_MS); - } catch (InterruptedException e) - { - Thread.currentThread().interrupt(); - throw new IllegalStateException( - "Interrupted while waiting for Retina readiness", e); - } - } - } } diff --git a/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/turbo/QueryScheduleServer.java b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/turbo/QueryScheduleServer.java index 5731631720..a8144b17e3 100644 --- a/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/turbo/QueryScheduleServer.java +++ b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/turbo/QueryScheduleServer.java @@ -35,7 +35,7 @@ public class QueryScheduleServer implements Server { private static final Logger log = LogManager.getLogger(QueryScheduleServer.class); - private boolean running = false; + private volatile boolean running = false; private final io.grpc.Server rpcServer; public QueryScheduleServer(int port) diff --git a/pixels-daemon/src/test/java/io/pixelsdb/pixels/daemon/TestServerContainer.java b/pixels-daemon/src/test/java/io/pixelsdb/pixels/daemon/TestServerContainer.java new file mode 100644 index 0000000000..e644334d3b --- /dev/null +++ b/pixels-daemon/src/test/java/io/pixelsdb/pixels/daemon/TestServerContainer.java @@ -0,0 +1,819 @@ +/* + * Copyright 2026 PixelsDB. + * + * This file is part of Pixels. + * + * Pixels is free software: you can redistribute it and/or modify + * it under the terms of the Affero GNU General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * Pixels is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Affero GNU General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public + * License along with Pixels. If not, see + * . + */ +package io.pixelsdb.pixels.daemon; + +import io.pixelsdb.pixels.common.server.Server; +import io.pixelsdb.pixels.daemon.exception.NoSuchServerException; +import org.junit.Test; + +import java.util.Collections; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +public class TestServerContainer +{ + private static final long WAIT_TIMEOUT_SECONDS = 2; + private static final long NO_EVENT_TIMEOUT_MILLIS = 100; + + @Test + public void testAddServerRegistersStartsAndShutsDownServer() throws Exception + { + BlockingServer server = new BlockingServer(true); + ServerContainer container = new ServerContainer(); + try + { + container.addServer("managed", server); + + assertTrue(server.started.await(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + assertEquals(Collections.singletonList("managed"), container.getServerNames()); + assertTrue(container.checkServer("managed")); + } + finally + { + stopBlockingServer(container, "managed", server); + } + assertEquals(1, server.shutdownCount.get()); + assertFalse(container.checkServer("managed")); + } + + @Test + public void testStartServerDoesNotDuplicateActiveLifecycleThread() throws Exception + { + BlockingServer server = new BlockingServer(false); + ServerContainer container = new ServerContainer(); + try + { + container.addServer("slow", server); + assertTrue(server.started.await(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + + container.startServer("slow"); + + assertFalse(server.secondRun.await( + NO_EVENT_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)); + assertEquals(1, server.runCount.get()); + } + finally + { + stopBlockingServer(container, "slow", server); + } + } + + @Test + public void testStartupChecksRunInOrderBeforeServer() throws Exception + { + BlockingServer server = new BlockingServer(false); + BlockingStartupCheck firstCheck = new BlockingStartupCheck("first check"); + BlockingStartupCheck secondCheck = new BlockingStartupCheck("second check"); + ServerContainer container = new ServerContainer(); + try + { + container.addServer("checked", server, firstCheck, secondCheck); + + assertTrue(firstCheck.entered.await(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + assertFalse(secondCheck.entered.await( + NO_EVENT_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)); + + container.startServer("checked"); + firstCheck.allowReady(); + + assertTrue(secondCheck.entered.await(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + assertFalse(server.started.await(NO_EVENT_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)); + + container.startServer("checked"); + secondCheck.allowReady(); + + assertTrue(server.started.await(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + assertEquals(1, server.runCount.get()); + assertFalse(server.secondRun.await( + NO_EVENT_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)); + } + finally + { + firstCheck.allowReady(); + secondCheck.allowReady(); + stopBlockingServer(container, "checked", server); + } + } + + @Test + public void testStartupCheckFailurePreventsServerStartup() throws Exception + { + BlockingServer server = new BlockingServer(false); + FailingStartupCheck startupCheck = new FailingStartupCheck(); + ServerContainer container = new ServerContainer(); + container.addServer("failing", server, startupCheck); + + assertTrue(startupCheck.executed.await(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + awaitLifecycleStopped(container, "failing"); + + assertEquals(0, server.runCount.get()); + assertEquals(1L, server.started.getCount()); + } + + @Test + public void testShutdownInterruptsPendingStartupCheck() throws Exception + { + BlockingServer server = new BlockingServer(false); + BlockingStartupCheck startupCheck = new BlockingStartupCheck("blocking check"); + ServerContainer container = new ServerContainer(); + try + { + container.addServer("waiting", server, startupCheck); + assertTrue(startupCheck.entered.await(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + + container.shutdownAll(); + + assertTrue(startupCheck.interrupted.await(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + awaitLifecycleStopped(container, "waiting"); + assertEquals(0, server.runCount.get()); + container.startServer("waiting"); + assertFalse(server.started.await( + NO_EVENT_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)); + } + finally + { + container.shutdownAll(); + startupCheck.allowReady(); + server.release(); + awaitLifecycleStopped(container, "waiting"); + } + } + + @Test + public void testRunningServerIsNotRestartedAfterRunReturns() throws Exception + { + AsyncServer server = new AsyncServer(); + ServerContainer container = new ServerContainer(); + try + { + container.addServer("async", server); + assertTrue(server.started.await(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + awaitLifecycleStopped(container, "async"); + + container.startServer("async"); + + assertFalse(server.restarted.await( + NO_EVENT_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)); + assertEquals(1, server.runCount.get()); + assertTrue(server.isRunning()); + } + finally + { + container.shutdownAll(); + } + assertEquals(1, server.shutdownCount.get()); + assertFalse(server.isRunning()); + } + + @Test + public void testStoppedServerCanBeRestarted() throws Exception + { + RestartableServer server = new RestartableServer(); + ServerContainer container = new ServerContainer(); + + container.addServer("restartable", server); + assertTrue(server.firstRun.await(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + awaitLifecycleStopped(container, "restartable"); + + container.startServer("restartable"); + + assertTrue(server.secondRun.await(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + awaitLifecycleStopped(container, "restartable"); + assertEquals(2, server.runCount.get()); + } + + @Test + public void testShutdownPreventsFurtherRestart() throws Exception + { + RestartableServer server = new RestartableServer(); + ServerContainer container = new ServerContainer(); + + container.addServer("restartable", server); + assertTrue(server.firstRun.await(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + awaitLifecycleStopped(container, "restartable"); + + container.shutdownAll(); + container.startServer("restartable"); + + assertFalse(server.secondRun.await( + NO_EVENT_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)); + assertEquals(1, server.runCount.get()); + } + + @Test + public void testShutdownAllStopsEveryServer() throws Exception + { + BlockingServer first = new BlockingServer(true); + BlockingServer second = new BlockingServer(true); + ServerContainer container = new ServerContainer(); + + container.addServer("first", first); + container.addServer("second", second); + assertTrue(first.started.await(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + assertTrue(second.started.await(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + + container.shutdownAll(); + + assertTrue(container.awaitTermination(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + assertEquals(1, first.shutdownCount.get()); + assertEquals(1, second.shutdownCount.get()); + } + + @Test + public void testShutdownAllIsIdempotent() throws Exception + { + AsyncServer server = new AsyncServer(); + ServerContainer container = new ServerContainer(); + container.addServer("async", server); + assertTrue(server.started.await(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + awaitLifecycleStopped(container, "async"); + + container.shutdownAll(); + container.shutdownAll(); + + assertTrue(container.awaitTermination(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + assertEquals(1, server.shutdownCount.get()); + } + + @Test + public void testShutdownAllContinuesAfterServerFailure() throws Exception + { + FailingShutdownServer failing = new FailingShutdownServer(); + BlockingServer healthy = new BlockingServer(true); + ServerContainer container = new ServerContainer(); + container.addServer("failing", failing); + container.addServer("healthy", healthy); + assertTrue(failing.started.await(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + assertTrue(healthy.started.await(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + + container.shutdownAll(); + + assertTrue(container.awaitTermination(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + assertEquals(1, failing.shutdownCount.get()); + assertEquals(1, healthy.shutdownCount.get()); + } + + @Test + public void testAwaitTerminationTimesOutForRunningAsyncServer() throws Exception + { + UnstoppableAsyncServer server = new UnstoppableAsyncServer(); + ServerContainer container = new ServerContainer(); + container.addServer("unstoppable", server); + assertTrue(server.started.await(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + awaitLifecycleStopped(container, "unstoppable"); + + container.shutdownAll(); + + assertFalse(container.awaitTermination( + NO_EVENT_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)); + assertEquals(1, server.shutdownCount.get()); + } + + @Test + public void testLateStartingServerIsShutdownAfterGateCloses() throws Exception + { + LateStartingServer server = new LateStartingServer(); + ServerContainer container = new ServerContainer(); + container.addServer("late", server); + assertTrue(server.runEntered.await(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + + try + { + container.shutdownAll(); + assertEquals(0, server.shutdownCount.get()); + + server.allowRunning(); + assertTrue(server.runningPublished.await( + WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + awaitLifecycleStopped(container, "late"); + + assertTrue(container.awaitTermination(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + assertEquals(1, server.shutdownCount.get()); + assertFalse(server.isRunning()); + } + finally + { + server.allowRunning(); + container.shutdownAll(); + container.awaitTermination(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS); + } + } + + @Test + public void testConcurrentStartCannotRestartAfterShutdownBegins() throws Exception + { + RestartableServer restartable = new RestartableServer(); + BlockingShutdownServer blocking = new BlockingShutdownServer(); + ServerContainer container = new ServerContainer(); + container.addServer("restartable", restartable); + container.addServer("blocking", blocking); + assertTrue(restartable.firstRun.await(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + assertTrue(blocking.started.await(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + awaitLifecycleStopped(container, "restartable"); + + Thread shutdownThread = new Thread(container::shutdownAll); + shutdownThread.start(); + try + { + assertTrue(blocking.shutdownEntered.await( + WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + + container.startServer("restartable"); + + assertFalse(restartable.secondRun.await( + NO_EVENT_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)); + } + finally + { + blocking.allowShutdown(); + shutdownThread.join(TimeUnit.SECONDS.toMillis(WAIT_TIMEOUT_SECONDS)); + } + assertFalse("shutdown thread did not stop", shutdownThread.isAlive()); + assertTrue(container.awaitTermination(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + assertEquals(1, restartable.runCount.get()); + } + + @Test + public void testAddServerIsRejectedAfterShutdown() + { + ServerContainer container = new ServerContainer(); + container.shutdownAll(); + + try + { + container.addServer("late", new RestartableServer()); + fail("server registration after shutdown should fail"); + } + catch (IllegalStateException expected) + { + assertEquals("server container is shutting down", expected.getMessage()); + } + } + + @Test + public void testUnknownServerOperationsFail() throws Exception + { + ServerContainer container = new ServerContainer(); + + assertNoSuchServer(() -> container.startServer("missing")); + assertNoSuchServer(() -> container.checkServer("missing")); + } + + @Test + public void testDuplicateServerRegistrationIsRejected() throws Exception + { + BlockingServer first = new BlockingServer(false); + BlockingServer duplicate = new BlockingServer(false); + ServerContainer container = new ServerContainer(); + try + { + container.addServer("duplicate", first); + assertTrue(first.started.await(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + + try + { + container.addServer("duplicate", duplicate); + fail("duplicate server registration should fail"); + } + catch (IllegalArgumentException expected) + { + assertEquals("server is already registered: duplicate", expected.getMessage()); + } + + assertEquals(1, first.runCount.get()); + assertEquals(0, duplicate.runCount.get()); + } + finally + { + stopBlockingServer(container, "duplicate", first); + } + } + + private static void stopBlockingServer( + ServerContainer container, String name, BlockingServer server) throws Exception + { + try + { + container.shutdownAll(); + } + finally + { + server.release(); + } + if (server.started.getCount() == 0) + { + assertTrue(server.stopped.await(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + } + awaitLifecycleStopped(container, name); + } + + private static void awaitLifecycleStopped(ServerContainer container, String name) + throws Exception + { + long deadline = System.nanoTime() + + TimeUnit.SECONDS.toNanos(WAIT_TIMEOUT_SECONDS); + while (container.checkServer(name) && System.nanoTime() < deadline) + { + TimeUnit.MILLISECONDS.sleep(10); + } + assertFalse("server lifecycle thread did not stop", container.checkServer(name)); + } + + private static void assertNoSuchServer(ServerOperation operation) throws Exception + { + try + { + operation.run(); + fail("operation on an unknown server should fail"); + } + catch (NoSuchServerException expected) + { + // Expected. + } + } + + private interface ServerOperation + { + void run() throws NoSuchServerException; + } + + private static final class BlockingStartupCheck implements StartupCheck + { + private final String description; + private final CountDownLatch entered = new CountDownLatch(1); + private final CountDownLatch ready = new CountDownLatch(1); + private final CountDownLatch interrupted = new CountDownLatch(1); + + private BlockingStartupCheck(String description) + { + this.description = description; + } + + @Override + public String getDescription() + { + return description; + } + + @Override + public void awaitReady(long deadlineNanos) throws InterruptedException + { + entered.countDown(); + try + { + long remainingNanos = deadlineNanos - System.nanoTime(); + if (remainingNanos <= 0 + || !ready.await(remainingNanos, TimeUnit.NANOSECONDS)) + { + throw new IllegalStateException(description + " timed out"); + } + } + catch (InterruptedException e) + { + interrupted.countDown(); + throw e; + } + } + + private void allowReady() + { + ready.countDown(); + } + } + + private static final class FailingStartupCheck implements StartupCheck + { + private final CountDownLatch executed = new CountDownLatch(1); + + @Override + public String getDescription() + { + return "failing startup check"; + } + + @Override + public void awaitReady(long deadlineNanos) + { + executed.countDown(); + throw new IllegalStateException("test startup check failed"); + } + } + + private static final class BlockingServer implements Server + { + private final boolean reportRunning; + private final CountDownLatch started = new CountDownLatch(1); + private final CountDownLatch stopped = new CountDownLatch(1); + private final CountDownLatch stop = new CountDownLatch(1); + private final CountDownLatch secondRun = new CountDownLatch(1); + private final AtomicInteger runCount = new AtomicInteger(); + private final AtomicInteger shutdownCount = new AtomicInteger(); + private volatile boolean running; + + private BlockingServer(boolean reportRunning) + { + this.reportRunning = reportRunning; + } + + @Override + public boolean isRunning() + { + return reportRunning && running; + } + + @Override + public void run() + { + if (runCount.incrementAndGet() > 1) + { + secondRun.countDown(); + } + running = true; + started.countDown(); + try + { + stop.await(); + } + catch (InterruptedException e) + { + Thread.currentThread().interrupt(); + } + finally + { + running = false; + stopped.countDown(); + } + } + + @Override + public void shutdown() + { + shutdownCount.incrementAndGet(); + stop.countDown(); + } + + private void release() + { + stop.countDown(); + } + } + + private static final class AsyncServer implements Server + { + private final CountDownLatch started = new CountDownLatch(1); + private final CountDownLatch restarted = new CountDownLatch(1); + private final AtomicInteger runCount = new AtomicInteger(); + private final AtomicInteger shutdownCount = new AtomicInteger(); + private volatile boolean running; + + @Override + public boolean isRunning() + { + return running; + } + + @Override + public void run() + { + if (runCount.incrementAndGet() > 1) + { + restarted.countDown(); + } + running = true; + started.countDown(); + } + + @Override + public void shutdown() + { + shutdownCount.incrementAndGet(); + running = false; + } + } + + private static final class FailingShutdownServer implements Server + { + private final CountDownLatch started = new CountDownLatch(1); + private final CountDownLatch stop = new CountDownLatch(1); + private final AtomicInteger shutdownCount = new AtomicInteger(); + private volatile boolean running; + + @Override + public boolean isRunning() + { + return running; + } + + @Override + public void run() + { + running = true; + started.countDown(); + try + { + stop.await(); + } + catch (InterruptedException e) + { + Thread.currentThread().interrupt(); + } + finally + { + running = false; + } + } + + @Override + public void shutdown() + { + shutdownCount.incrementAndGet(); + throw new IllegalStateException("expected shutdown failure"); + } + } + + private static final class UnstoppableAsyncServer implements Server + { + private final CountDownLatch started = new CountDownLatch(1); + private final AtomicInteger shutdownCount = new AtomicInteger(); + private volatile boolean running; + + @Override + public boolean isRunning() + { + return running; + } + + @Override + public void run() + { + running = true; + started.countDown(); + } + + @Override + public void shutdown() + { + shutdownCount.incrementAndGet(); + } + } + + private static final class LateStartingServer implements Server + { + private final CountDownLatch runEntered = new CountDownLatch(1); + private final CountDownLatch runningPublished = new CountDownLatch(1); + private final CountDownLatch continueRunning = new CountDownLatch(1); + private final AtomicInteger shutdownCount = new AtomicInteger(); + private volatile boolean running; + + @Override + public boolean isRunning() + { + return running; + } + + @Override + public void run() + { + runEntered.countDown(); + boolean allowedToRun = false; + while (!allowedToRun) + { + try + { + continueRunning.await(); + allowedToRun = true; + } + catch (InterruptedException ignored) + { + // Deliberately ignore interruption to reproduce an in-flight start. + } + } + running = true; + runningPublished.countDown(); + } + + @Override + public void shutdown() + { + shutdownCount.incrementAndGet(); + running = false; + } + + private void allowRunning() + { + continueRunning.countDown(); + } + } + + private static final class BlockingShutdownServer implements Server + { + private final CountDownLatch started = new CountDownLatch(1); + private final CountDownLatch shutdownEntered = new CountDownLatch(1); + private final CountDownLatch continueShutdown = new CountDownLatch(1); + private final CountDownLatch stop = new CountDownLatch(1); + private volatile boolean running; + + @Override + public boolean isRunning() + { + return running; + } + + @Override + public void run() + { + running = true; + started.countDown(); + try + { + stop.await(); + } + catch (InterruptedException e) + { + Thread.currentThread().interrupt(); + } + finally + { + running = false; + } + } + + @Override + public void shutdown() + { + shutdownEntered.countDown(); + try + { + continueShutdown.await(); + } + catch (InterruptedException e) + { + Thread.currentThread().interrupt(); + } + finally + { + stop.countDown(); + } + } + + private void allowShutdown() + { + continueShutdown.countDown(); + } + } + + private static final class RestartableServer implements Server + { + private final CountDownLatch firstRun = new CountDownLatch(1); + private final CountDownLatch secondRun = new CountDownLatch(1); + private final AtomicInteger runCount = new AtomicInteger(); + + @Override + public boolean isRunning() + { + return false; + } + + @Override + public void run() + { + int currentRun = runCount.incrementAndGet(); + if (currentRun == 1) + { + firstRun.countDown(); + } + else if (currentRun == 2) + { + secondRun.countDown(); + } + } + + @Override + public void shutdown() + { + } + } +} diff --git a/pixels-retina/src/main/java/io/pixelsdb/pixels/retina/ObjectStorageManager.java b/pixels-retina/src/main/java/io/pixelsdb/pixels/retina/ObjectStorageManager.java index 20d7114703..9a13a8bf2e 100644 --- a/pixels-retina/src/main/java/io/pixelsdb/pixels/retina/ObjectStorageManager.java +++ b/pixels-retina/src/main/java/io/pixelsdb/pixels/retina/ObjectStorageManager.java @@ -49,15 +49,21 @@ private ObjectStorageManager() throws RetinaException } this.path = folder; - String storageScheme = config.getProperty("retina.buffer.object.storage.scheme"); try { - if (!Storage.Scheme.minio.equals(storageScheme) && !Storage.Scheme.s3.equals(storageScheme)) + Storage.Scheme scheme = Storage.Scheme.fromPath(folder); + if (scheme == null) + { + throw new RetinaException("retina.buffer.object.storage.folder must include a storage scheme prefix, " + + "e.g., s3://bucket/path/"); + } + if (!Storage.Scheme.minio.equals(scheme) && !Storage.Scheme.s3.equals(scheme) + && !Storage.Scheme.cos.equals(scheme)) { logger.warn("ObjectStorageManager is configured with non-standard storage scheme '{}'; " + - "expected production schemes are 's3' or 'minio'", storageScheme); + "expected production schemes are 's3', 'minio' or 'cos'", scheme); } - this.storage = StorageFactory.Instance().getStorage(storageScheme); + this.storage = StorageFactory.Instance().getStorage(folder); } catch (IOException e) { throw new RetinaException("Failed to get storage", e); diff --git a/skills/README.md b/skills/README.md index 6f8114566b..f9b2ff20cb 100644 --- a/skills/README.md +++ b/skills/README.md @@ -10,6 +10,8 @@ Pixels; users explicitly install a skill with `skills/install.sh`. - `skills//`: One independent skill development directory. - `skills//skill.yaml`: Skill metadata, script map, and phase list. - `skills//codex/SKILL.md`: Codex skill entrypoint. +- `skills//cursor/SKILL.md`: Cursor skill entrypoint (falls back to + `codex/SKILL.md` when absent). - `skills//claude/agent.md`: Claude Code agent entrypoint, when supported. - `skills//scripts/`: Runtime scripts bundled with the installed skill. - `skills//scripts/lib/`: Shell helpers sourced by that skill's own scripts. @@ -19,8 +21,8 @@ Pixels; users explicitly install a skill with `skills/install.sh`. ### Commands - `./skills/list.sh`: List skill directories that contain `skill.yaml`. -- `./skills/install.sh --skill --tool --scope `: Install a skill for the selected tool and scope. -- `./skills/uninstall.sh --skill --tool --scope `: Remove an installed skill for the selected tool and scope. +- `./skills/install.sh --skill --tool --scope `: Install a skill for the selected tool and scope. +- `./skills/uninstall.sh --skill --tool --scope `: Remove an installed skill for the selected tool and scope. `--agent ` is accepted by install/uninstall as a deprecated alias for `--skill ` during the migration from the old `agents/` layout. @@ -41,6 +43,23 @@ discovery path: $HOME/.agents/skills// ``` +Cursor project-scope installs copy the selected skill into Cursor's project +skill discovery path: + +```text +/.cursor/skills// +``` + +Cursor global-scope installs copy the selected skill into Cursor's personal +skill discovery path: + +```text +$HOME/.cursor/skills// +``` + +Cursor installs never write into `.cursor/skills-cursor/`, which is reserved +for Cursor's built-in skills. + Claude project/global installs copy the Claude markdown agent to the matching `.claude/agents/` directory and copy scripts/references to a companion assets directory: @@ -58,4 +77,6 @@ installed skill bundle. The pixels-install scripts default to: $HOME/.agents/state/pixels-install/ ``` -Override with `STATE_DIR=` when a different state location is required. +That state location is shared across Codex and Cursor installs of the same +skill so progress/config stay in one place. Override with `STATE_DIR=` +when a different state location is required. diff --git a/skills/install.sh b/skills/install.sh index f21dc07055..b8590a3866 100755 --- a/skills/install.sh +++ b/skills/install.sh @@ -23,12 +23,13 @@ available_skills() { usage() { cat < --tool --scope + $0 --skill --tool --scope Required arguments: --skill Skill directory name under skills/. --agent is accepted as a deprecated alias. - --tool Target AI tool. + --tool + Target AI tool. --scope project: install into this repository. global: install into the current user's tool config. @@ -37,6 +38,8 @@ Examples: $0 --skill pixels-install --tool codex --scope project $0 --skill pixels-install --tool codex --scope global $0 --skill pixels-install --tool claude --scope project + $0 --skill pixels-install --tool cursor --scope project + $0 --skill pixels-install --tool cursor --scope global Available skills: $(available_skills) @@ -107,7 +110,7 @@ parse_args() { done [ -n "$SKILL" ] || ERRORS+=("missing --skill ") - [ -n "$TOOL" ] || ERRORS+=("missing --tool ") + [ -n "$TOOL" ] || ERRORS+=("missing --tool ") [ -n "$SCOPE" ] || ERRORS+=("missing --scope ") if [ -n "$SKILL" ]; then @@ -116,8 +119,8 @@ parse_args() { if [ -n "$TOOL" ]; then case "$TOOL" in - codex|claude) ;; - *) ERRORS+=("--tool must be codex or claude") ;; + codex|claude|cursor) ;; + *) ERRORS+=("--tool must be codex, claude, or cursor") ;; esac fi @@ -180,6 +183,14 @@ codex_target_dir() { fi } +cursor_target_dir() { + if [ "$SCOPE" = "project" ]; then + printf '%s/.cursor/skills/%s\n' "$(project_root)" "$SKILL" + else + printf '%s/.cursor/skills/%s\n' "$HOME" "$SKILL" + fi +} + claude_agent_target() { if [ "$SCOPE" = "project" ]; then printf '%s/.claude/agents/%s.md\n' "$(project_root)" "$SKILL" @@ -212,6 +223,33 @@ install_codex() { echo "Installed Codex skill '$SKILL' to $target_dir" } +install_cursor() { + local source_dir target_dir source_skill + source_dir="$(skill_dir)" + source_skill="$source_dir/cursor/SKILL.md" + target_dir="$(cursor_target_dir)" + + # Prefer a Cursor-specific entrypoint; fall back to the Codex SKILL.md, + # which already uses Cursor-compatible YAML frontmatter. + if [ ! -f "$source_skill" ]; then + source_skill="$source_dir/codex/SKILL.md" + fi + [ -f "$source_skill" ] || fail_with_usage "Cursor SKILL.md not found: $source_dir/cursor/SKILL.md (or codex/SKILL.md)" + + case "$target_dir" in + */.cursor/skills-cursor/*|*/.cursor/skills-cursor) + fail_with_usage "refusing to install into .cursor/skills-cursor (reserved for Cursor built-in skills)" + ;; + esac + + rm -rf "$target_dir" + mkdir -p "$target_dir" + cp "$source_skill" "$target_dir/SKILL.md" + copy_common_assets "$source_dir" "$target_dir" + + echo "Installed Cursor skill '$SKILL' to $target_dir" +} + install_claude() { local source_dir source_agent target_agent target_assets source_dir="$(skill_dir)" @@ -272,4 +310,5 @@ ensure_skill_exists case "$TOOL" in codex) install_codex ;; claude) install_claude ;; + cursor) install_cursor ;; esac diff --git a/skills/pixels-install/claude/agent.md b/skills/pixels-install/claude/agent.md index 57bde4a468..6633f9423b 100644 --- a/skills/pixels-install/claude/agent.md +++ b/skills/pixels-install/claude/agent.md @@ -30,8 +30,17 @@ Use these helper scripts when they directly fit the current environment and the - `install_trino.sh`: installs Trino (per the [466 deployment docs](https://trino.io/docs/466/installation/deployment.html)) on the **current node only** — run it once per node in the cluster, it does not SSH anywhere itself (use `install_trino_cluster.sh` to fan it out across the cluster instead of running it by hand on each node). It reads `trino-deployment.env`, validates this node's role (`coordinator` or `worker`) and the coordinator host, and confirms the role plus install/data paths before writing files; in non-interactive mode, set `CONFIRM_TRINO_INSTALL=true` only after the user confirms the target. Downloads `trino-server-.tar.gz` (default 466) from Maven Central into the confirmed install parent with the confirmed home symlink for version-switch-friendly installs. Writes `etc/node.properties` (`node.data-dir` under the confirmed data dir, outside the versioned install dir so it survives version switches; `node.id` generated once and preserved across re-runs), `etc/jvm.config` (official defaults plus the two `--add-opens` flags `pixels-trino` requires), `etc/log.properties`, and `etc/config.properties` — `discovery.uri` always points at the coordinator's real IP from `trino-deployment.env`, never `localhost`; this node's coordinator/worker role is read from `trino-deployment.env` or inferred from this host's own IPs, overridable with `TRINO_ROLE`. For `pixelsdb/pixels-trino`, either clones/builds locally (requires Pixels already `mvn install`ed locally and JDK 23) or consumes prebuilt `PIXELS_TRINO_CONNECTOR_ZIP` and `PIXELS_TRINO_LISTENER_ZIP` so remote Trino nodes do not need a Pixels checkout or Maven build. Installs both documented plugins by directly unzipping their zips under `plugin/` and preserving the zip's versioned top-level directory names, e.g. `pixels-trino-connector-0.2.0-SNAPSHOT` and `pixels-trino-listener-0.2.0-SNAPSHOT`; do not rename them to unversioned stable directories. Writes Trino's catalog file `etc/catalog/pixels.properties` with only catalog properties (`connector.name=pixels`, `cloud.function.switch=off`, `clean.intermediate.result=true` by default). This catalog file is **not** the same thing as Pixels' client/runtime config file `$PIXELS_HOME/etc/pixels.properties`; never copy Pixels runtime properties such as `metadata.server.host` or `etcd.hosts` into the Trino catalog. Also prepares the Trino-side lightweight Pixels client home (`TRINO_PIXELS_HOME`, default `$HOME/opt/pixels`, not `$HOME/opt/pixels/etc`) with `etc/pixels.properties` and `logs/`, then writes `export PIXELS_HOME=...` and `export PIXELS_CONFIG=$PIXELS_HOME/etc/pixels.properties` directly into the current user's bash/zsh profile (and mirrors them into `STATE_DIR/toolchain.env` for later helper-script processes). It does **not** create `~/.pixels-trino-env.sh`. Use `TRINO_PIXELS_CONFIG_SOURCE` to copy a known-good Pixels config into that client home, and use `TRINO_PIXELS_METADATA_SERVER_HOST`, `TRINO_PIXELS_TRANS_SERVER_HOST`, `TRINO_PIXELS_QUERY_SCHEDULE_SERVER_HOST`, and `TRINO_PIXELS_ETCD_HOSTS` to rewrite endpoints so Trino nodes do not fall back to `localhost`. Writes `etc/event-listener.properties` for the listener and defaults its log directory to a Trino-side path under the install parent unless `PIXELS_TRINO_LISTENER_LOG_DIR` is set. Best-effort installs `trino-cli` into `bin/trino`. - `install_trino_shell_helpers.sh`: optional, and **always asks first**, because it bakes a specific list of remote hostnames (from `trino-deployment.env`) into a shell profile and assumes passwordless SSH between them is already set up. By default it installs only `start_trino_cluster`/`stop_trino_cluster`/`restart_trino_cluster`/`trino_cli` on the Trino coordinator recorded in `trino-deployment.env`; when invoked from another host, it copies a minimal installer over SSH and writes the coordinator user's `~/.trino-shell-helpers.sh` plus shell-profile source line there. In the generated functions, the coordinator is always operated locally and every worker over SSH. `stop_trino_cluster` stops workers before the coordinator; `restart_trino_cluster` runs stop then start. The generated start/stop/restart/client helpers explicitly export `PIXELS_HOME` and `PIXELS_CONFIG` before invoking Trino locally or over SSH, without wrapping or replacing Trino's `bin/launcher`. Set `TRINO_SHELL_HELPERS_TARGET=local` only when the user explicitly wants the functions on the current host. Leave `INSTALL_TRINO_SHELL_HELPERS` unset to be prompted `[y/N]`; set `true`/`false` to answer non-interactively once the user has actually agreed, or `ASSUME_YES=true`. +All executable entrypoints in this skill are Bash programs. Invoke them directly +so their `#!/usr/bin/env bash` shebang is honored; do not run them with `zsh` +or `source` them into the current shell. Bash must be installed on every +deployment node even when that account's login shell is Zsh. Each entrypoint +validates the deployment account's configured login shell from passwd/NSS and +accepts only Bash or Zsh. Profile exports go to `.bashrc` or `.zshrc` +accordingly, while the generated Pixels and Trino helper files are checked in +both Bash and Zsh syntax before they are sourced. + All of these scripts anchor paths on `$HOME` (e.g. `~/opt/...`) and detect the -current login shell (`$SHELL`) to pick `~/.bashrc` or `~/.zshrc` before +current account's login shell from passwd/NSS to pick `~/.bashrc` or `~/.zshrc` before persisting any `export`. Never assume the deployment user is named `pixels` or `ubuntu`, and never write environment variables into a global file such as `/etc/environment` — `skills/pixels-install/scripts/lib/shell_env.sh` diff --git a/skills/pixels-install/codex/SKILL.md b/skills/pixels-install/codex/SKILL.md index 4fc84de608..602913e028 100644 --- a/skills/pixels-install/codex/SKILL.md +++ b/skills/pixels-install/codex/SKILL.md @@ -38,8 +38,17 @@ Use these helpers when they directly fit the current environment and the user-ap - `install_trino.sh`: installs Trino (per the [466 deployment docs](https://trino.io/docs/466/installation/deployment.html)) on the **current node only** — run it once per node in the cluster, it does not SSH anywhere itself (use `install_trino_cluster.sh` to fan it out across the cluster instead of running it by hand on each node). It reads `trino-deployment.env`, validates this node's role (`coordinator` or `worker`) and the coordinator host, and confirms the role plus install/data paths before writing files; in non-interactive mode, set `CONFIRM_TRINO_INSTALL=true` only after the user confirms the target. Downloads `trino-server-.tar.gz` (default 466) from Maven Central into the confirmed install parent with the confirmed home symlink for version-switch-friendly installs. Writes `etc/node.properties` (`node.data-dir` under the confirmed data dir, outside the versioned install dir so it survives version switches; `node.id` generated once and preserved across re-runs), `etc/jvm.config` (official defaults plus the two `--add-opens` flags `pixels-trino` requires), `etc/log.properties`, and `etc/config.properties` — `discovery.uri` always points at the coordinator's real IP from `trino-deployment.env`, never `localhost`; this node's coordinator/worker role is read from `trino-deployment.env` or inferred from this host's own IPs, overridable with `TRINO_ROLE`. For `pixelsdb/pixels-trino`, either clones/builds locally (requires Pixels already `mvn install`ed locally and JDK 23) or consumes prebuilt `PIXELS_TRINO_CONNECTOR_ZIP` and `PIXELS_TRINO_LISTENER_ZIP` so remote Trino nodes do not need a Pixels checkout or Maven build. Installs both documented plugins by directly unzipping their zips under `plugin/` and preserving the zip's versioned top-level directory names, e.g. `pixels-trino-connector-0.2.0-SNAPSHOT` and `pixels-trino-listener-0.2.0-SNAPSHOT`; do not rename them to unversioned stable directories. Writes Trino's catalog file `etc/catalog/pixels.properties` with only catalog properties (`connector.name=pixels`, `cloud.function.switch=off`, `clean.intermediate.result=true` by default). This catalog file is **not** the same thing as Pixels' client/runtime config file `$PIXELS_HOME/etc/pixels.properties`; never copy Pixels runtime properties such as `metadata.server.host` or `etcd.hosts` into the Trino catalog. Also prepares the Trino-side lightweight Pixels client home (`TRINO_PIXELS_HOME`, default `$HOME/opt/pixels`, not `$HOME/opt/pixels/etc`) with `etc/pixels.properties` and `logs/`, then writes `export PIXELS_HOME=...` and `export PIXELS_CONFIG=$PIXELS_HOME/etc/pixels.properties` directly into the current user's bash/zsh profile (and mirrors them into `STATE_DIR/toolchain.env` for later helper-script processes). It does **not** create `~/.pixels-trino-env.sh`. Use `TRINO_PIXELS_CONFIG_SOURCE` to copy a known-good Pixels config into that client home, and use `TRINO_PIXELS_METADATA_SERVER_HOST`, `TRINO_PIXELS_TRANS_SERVER_HOST`, `TRINO_PIXELS_QUERY_SCHEDULE_SERVER_HOST`, and `TRINO_PIXELS_ETCD_HOSTS` to rewrite endpoints so Trino nodes do not fall back to `localhost`. Writes `etc/event-listener.properties` for the listener and defaults its log directory to a Trino-side path under the install parent unless `PIXELS_TRINO_LISTENER_LOG_DIR` is set. Best-effort installs `trino-cli` into `bin/trino`. - `install_trino_shell_helpers.sh`: optional, and **always asks first**, because it bakes a specific list of remote hostnames (from `trino-deployment.env`) into a shell profile and assumes passwordless SSH between them is already set up. By default it installs only `start_trino_cluster`/`stop_trino_cluster`/`restart_trino_cluster`/`trino_cli` on the Trino coordinator recorded in `trino-deployment.env`; when invoked from another host, it copies a minimal installer over SSH and writes the coordinator user's `~/.trino-shell-helpers.sh` plus shell-profile source line there. In the generated functions, the coordinator is always operated locally and every worker over SSH. `stop_trino_cluster` stops workers before the coordinator; `restart_trino_cluster` runs stop then start. The generated start/stop/restart/client helpers explicitly export `PIXELS_HOME` and `PIXELS_CONFIG` before invoking Trino locally or over SSH, without wrapping or replacing Trino's `bin/launcher`. Set `TRINO_SHELL_HELPERS_TARGET=local` only when the user explicitly wants the functions on the current host. Leave `INSTALL_TRINO_SHELL_HELPERS` unset to be prompted `[y/N]`; set `true`/`false` to answer non-interactively once the user has actually agreed, or `ASSUME_YES=true`. +All executable entrypoints in this skill are Bash programs. Invoke them directly +so their `#!/usr/bin/env bash` shebang is honored; do not run them with `zsh` +or `source` them into the current shell. Bash must be installed on every +deployment node even when that account's login shell is Zsh. Each entrypoint +validates the deployment account's configured login shell from passwd/NSS and +accepts only Bash or Zsh. Profile exports go to `.bashrc` or `.zshrc` +accordingly, while the generated Pixels and Trino helper files are checked in +both Bash and Zsh syntax before they are sourced. + All of these scripts anchor paths on `$HOME` (e.g. `~/opt/...`) and detect the -current login shell (`$SHELL`) to pick `~/.bashrc` or `~/.zshrc` before +current account's login shell from passwd/NSS to pick `~/.bashrc` or `~/.zshrc` before persisting any `export`. Never assume the deployment user is named `pixels` or `ubuntu`, and never write environment variables into a global file such as `/etc/environment` — `scripts/lib/shell_env.sh` diff --git a/skills/pixels-install/cursor/SKILL.md b/skills/pixels-install/cursor/SKILL.md new file mode 100644 index 0000000000..fcb4de4c35 --- /dev/null +++ b/skills/pixels-install/cursor/SKILL.md @@ -0,0 +1,184 @@ +--- +name: pixels-install +description: Guide, execute, and verify a basic Pixels deployment from docs/INSTALL.md using bundled helper scripts. Use when the user asks to install, configure, resume, validate, or troubleshoot a basic Pixels deployment or optional Trino setup. +--- + +# pixels-install + +Help the user install a basic Pixels deployment according to `docs/INSTALL.md`. + +Treat `docs/INSTALL.md` as the source of truth for the installation flow. Use the bundled scripts only as helpers for deterministic, repeatable, or easy-to-misconfigure steps. Do not treat this skill as a one-shot unattended installer. + +## Resource Paths + +- Skill directory: use the directory containing this `SKILL.md`. +- Helper scripts: `scripts/`. +- Shared helper scripts: `shared-scripts/`. +- Runtime state: `STATE_DIR` when set; otherwise the scripts resolve it to `/.agents/state/pixels-install` for project installs (from either `.cursor/skills/` or `.agents/skills/`) or `$HOME/.agents/state/pixels-install` for global installs. +- Pixels source tree: `REPO_ROOT` when set; otherwise scripts try to discover the current Git repository root and fail with a clear message if they cannot. + +### Helper scripts + +Use these helpers when they directly fit the current environment and the user-approved goal: + +- `prepare_deployment.sh`: collect and confirm the Pixels deployment topology before any install writes to disk. With no node arguments it prompts for single-node vs. cluster, `PIXELS_HOME`, the coordinator node, whether the coordinator also runs a worker, and worker nodes; with arguments, use `--coordinator ` plus repeated `--worker <...>` (legacy `--node` still works). Writes `STATE_DIR/deployment.env` and optionally delegates SSH setup to `shared-scripts/setup_cluster.sh`. In non-interactive mode, set `CONFIRM_PIXELS_DEPLOYMENT=true` only after the user has reviewed the chosen topology and paths. +- `check_prerequisites.sh`: validate OS, architecture, memory, disk, ports, host resolution, privilege, and optional SSH reachability. Runs every check and exits with a structured `=== check_prerequisites result ===` summary (one `ok|warn|fail|skip : ` line per check, plus a final `summary: ok=N warn=N fail=N skip=N status=pass|fail` line) instead of stopping at the first failure, so a single run shows every problem at once. +- `install_jdk.sh`: install a Zulu OpenJDK build (default JDK 23) matching the server's CPU architecture (x86_64 or aarch64) under `~/opt`, and persist `JAVA_HOME` only into the current user's shell profile. Looks for a JDK that already satisfies `JDK_VERSION` first — checking `JAVA_HOME` if set, otherwise resolving whatever `java` is on `PATH` (e.g. an `apt`-installed JDK that never exported `JAVA_HOME`) — and if one is found, skips the download entirely and only fixes up the environment variable to point at it. Only downloads/installs a fresh Zulu build when nothing on the system satisfies the version requirement. Falls back to pointing at the manual `.deb` method in `docs/INSTALL.md` if the Azul metadata API lookup fails. +- `install_maven.sh`: install or validate Maven 3.8+ (default pinned version `3.9.8`; override with `MAVEN_VERSION` only if the user explicitly asks for a different one) when the current Maven is missing or incompatible with the selected JDK. Same "skip if already satisfied" logic as `install_jdk.sh`: if an existing `mvn` (from `apt`, a prior manual install, etc.) already meets the minimum version, it reuses that installation's home directory instead of downloading anything. Fresh installs go into `~/opt/apache-maven-` with a `~/opt/maven` symlink, never into `/opt`. +- `install_mysql.sh`: install MySQL, then interactively confirm the root password and the `pixels` metadata-DB-user password (default `password` for both, never applied silently), create the `pixels_metadata` database/user, and load `scripts/sql/metadata_schema.sql` from the Pixels source tree. It validates DB/user identifiers, quotes SQL values, avoids putting the root password in command-line arguments, and writes shell-quoted `STATE_DIR/deployment.secrets.env` (mode 600) so the confirmed credentials flow into `configure_pixels.sh` automatically. +- `install_etcd.sh`: install and optionally start the bundled etcd 3.3.4 package under `~/opt`. Always fixes the shipped `conf.yml`'s hardcoded `/home/ubuntu/...` `data-dir` to the real install path. By default (`ETCD_ALLOW_REMOTE=false`) it keeps etcd localhost-only. For a cluster, set `ETCD_ALLOW_REMOTE=true` only after confirming private networking/security-group rules; the script then requires `CONFIRM_ETCD_REMOTE_ACCESS=true`, `ASSUME_YES=true`, or an interactive confirmation before binding the client/peer listeners for remote access. Can also install and enable a `systemd` unit so etcd survives reboots and restarts on failure, but only after asking — it never installs this silently. Leave `INSTALL_ETCD_SYSTEMD_SERVICE` unset to be prompted interactively (`[y/N]`, defaults to no); set it to `true`/`false` to answer non-interactively once the user has actually agreed, or `ASSUME_YES=true` to answer yes to this and other yes/no prompts. Declining (or running non-interactively with no explicit answer) just skips the unit — etcd still starts, as a manually backgrounded process instead. Falls back the same way when `systemctl` isn't available at all. Restarts/re-enables the installed service automatically when the config actually changed. +- `build_install_pixels.sh`: build and install Pixels into the confirmed `PIXELS_HOME` from `deployment.env`, then add the MySQL JDBC connector. It prints the repo, deployment file, and install target before running; in non-interactive mode, set `CONFIRM_PIXELS_INSTALL=true` only after the user confirms the target. If you must bypass the prepared deployment file, pass `USE_DEPLOYMENT_FILE=false PIXELS_HOME=` explicitly. `AUTO_CONFIRM_INSTALL` (default `true`) auto-answers `install.sh`'s prompts only when that is safe; if `PIXELS_HOME/etc/pixels.properties` or `pixels-cpp.properties` already exist, it forces the interactive path instead, because `install.sh` would otherwise prompt to add/remove config options or overwrite `pixels-cpp.properties` outright — exactly what `Claude agent`/`SKILL.md` say not to do without the user's say-so. +- `configure_pixels.sh`: update `PIXELS_HOME/etc/pixels.properties` after database, etcd, host, port, topology, and path values are known. Automatically sources `deployment.env` and `deployment.secrets.env` (if present), so `PIXELS_HOME`, coordinator service hosts, worker names, and MySQL credentials stay in sync with what the earlier scripts confirmed. It writes `$PIXELS_HOME/etc/workers` from `PIXELS_WORKERS` by default. It fails instead of silently writing the documented default DB password unless `METADATA_DB_PASSWORD` is explicit or `ALLOW_DEFAULT_METADATA_PASSWORD=true` is set. +- `install_shell_helpers.sh`: optional convenience step that **asks first** before writing shell functions. By default it installs only `start_pixels`/`stop_pixels`/`restart_pixels` on the Pixels coordinator recorded in `deployment.env`; when invoked from another host, it copies a minimal installer over SSH and writes the coordinator user's `~/.pixels-shell-helpers.sh` plus shell-profile source line there. These wrap `$PIXELS_HOME/sbin/start-pixels.sh` and `$PIXELS_HOME/sbin/stop-pixels.sh`. Set `PIXELS_SHELL_HELPERS_TARGET=local` only when the user explicitly wants the functions on the current host. Leave `INSTALL_PIXELS_SHELL_HELPERS` unset to be prompted `[y/N]`; set `true`/`false` to answer non-interactively once the user has actually agreed, or `ASSUME_YES=true`. +- `smoke_test.sh`: verify the installed layout, configuration, Java (`CHECK_JAVA=true`), metadata access, etcd health, Pixels topology (`deployment.env` plus `$PIXELS_HOME/etc/workers`), core service ports (`CHECK_CORE_SERVICES=true`), `$PIXELS_HOME/logs` error patterns (`CHECK_PIXELS_LOGS=true`), and basic CLI behavior when applicable. It also checks Trino layout/catalog files when `CHECK_TRINO=true` and `trino-deployment.env` is present, can separately validate the lightweight Trino-side Pixels client config plus shell-profile `PIXELS_HOME`/`PIXELS_CONFIG` exports (`CHECK_TRINO_PIXELS_CLIENT=true`) without requiring a full Pixels runtime layout (`CHECK_PIXELS_LAYOUT=false`), can check Trino launcher status across the recorded cluster (`CHECK_TRINO_CLUSTER_STATUS=true`), checks that Trino logs did not fall back to classpath `pixels.properties` (`CHECK_TRINO_LOGS=true`), waits for `/v1/info` to report `starting=false` (`CHECK_TRINO_READY=true`), and can run `trino --catalog pixels --execute "SHOW SCHEMAS"` (`CHECK_TRINO_CLI=true`). `SHOW SCHEMAS` only needs to succeed; an empty Pixels catalog is expected before data is loaded. Same structured-summary behavior as `check_prerequisites.sh` above. +- `progress.sh`: records which of `skill.yaml`'s `phases` have actually completed, in `STATE_DIR/progress.env`. `progress.sh mark [note]` after a phase succeeds, `progress.sh show` to see what's already done, `progress.sh is-done ` to check one phase, `progress.sh reset` to start over. Exists so a session interrupted partway through a multi-phase install can resume from the actual state instead of guessing or re-running completed steps. +- `build_pixels_trino_artifacts.sh`: optional Trino-prep step. Builds `pixelsdb/pixels-trino` once on the local machine where Pixels modules have already been `mvn install`ed, copies `pixels-trino-connector-*.zip` and `pixels-trino-listener-*.zip` into `STATE_DIR/pixels-trino-artifacts/`, and writes `STATE_DIR/pixels-trino-artifacts.env` for `install_trino_cluster.sh`. In non-interactive mode, set `CONFIRM_PIXELS_TRINO_ARTIFACTS=true` only after the user confirms the checkout/artifact paths. +- `install_trino_cluster.sh`: run **on the coordinator** after `prepare_trino_cluster.sh`. It prints and confirms the Trino deployment file, version, install parent, home link, data dir, coordinator, coordinator scheduling mode, workers, Trino-side Pixels client home/config source, Pixels service endpoints, and any prebuilt pixels-trino plugin zips before installing; in non-interactive mode, set `CONFIRM_TRINO_CLUSTER_INSTALL=true` only after the user confirms the fan-out. Installs Trino locally for the coordinator's own role, then dispatches `install_trino.sh` on every worker **in parallel** over the coordinator -> worker passwordless SSH that `prepare_trino_cluster.sh` already set up, instead of requiring the user or skill to copy the same command and run it by hand once per node. Copies the current `STATE_DIR/trino-deployment.env` to each worker first (workers don't need a shared filesystem). If `PIXELS_TRINO_CONNECTOR_ZIP` and `PIXELS_TRINO_LISTENER_ZIP` are provided, or `STATE_DIR/pixels-trino-artifacts.env` exists from `build_pixels_trino_artifacts.sh`, it also copies a minimal installer plus those prebuilt plugin zips to each worker, so workers do not need a Pixels checkout or Maven build. It also copies the selected Trino-side Pixels client config (`TRINO_PIXELS_CONFIG_SOURCE`, defaulting to `$PIXELS_HOME/etc/pixels.properties` when available) and lets `install_trino.sh` rewrite service endpoints through `TRINO_PIXELS_*` variables. Remote Trino nodes receive Trino, the two pixels-trino plugin zips, Trino catalog/listener configs, and a lightweight `~/opt/pixels/etc/pixels.properties` client config — not a full Pixels installation. Emits the same structured per-node summary as `smoke_test.sh`/`check_prerequisites.sh`, with a log file per node under `STATE_DIR/logs/` for diagnosing any failure. +- `prepare_trino_cluster.sh`: collects and confirms the Trino topology and install locations: one coordinator, zero or more workers, whether the coordinator also accepts scheduled work, Trino version, HTTP port, install parent, home symlink/current path, and data directory. With no node arguments it prompts interactively; with arguments, use `--coordinator ` plus repeated `--worker <...>`. Writes `STATE_DIR/trino-deployment.env`. Optionally delegates to `shared-scripts/setup_cluster.sh` (`--setup-ssh true`) so the coordinator and every worker can passwordlessly SSH into each other, which the cluster shell helpers below depend on. In non-interactive mode, set `CONFIRM_TRINO_DEPLOYMENT=true` only after the user has reviewed the topology and paths. +- `install_trino.sh`: installs Trino (per the [466 deployment docs](https://trino.io/docs/466/installation/deployment.html)) on the **current node only** — run it once per node in the cluster, it does not SSH anywhere itself (use `install_trino_cluster.sh` to fan it out across the cluster instead of running it by hand on each node). It reads `trino-deployment.env`, validates this node's role (`coordinator` or `worker`) and the coordinator host, and confirms the role plus install/data paths before writing files; in non-interactive mode, set `CONFIRM_TRINO_INSTALL=true` only after the user confirms the target. Downloads `trino-server-.tar.gz` (default 466) from Maven Central into the confirmed install parent with the confirmed home symlink for version-switch-friendly installs. Writes `etc/node.properties` (`node.data-dir` under the confirmed data dir, outside the versioned install dir so it survives version switches; `node.id` generated once and preserved across re-runs), `etc/jvm.config` (official defaults plus the two `--add-opens` flags `pixels-trino` requires), `etc/log.properties`, and `etc/config.properties` — `discovery.uri` always points at the coordinator's real IP from `trino-deployment.env`, never `localhost`; this node's coordinator/worker role is read from `trino-deployment.env` or inferred from this host's own IPs, overridable with `TRINO_ROLE`. For `pixelsdb/pixels-trino`, either clones/builds locally (requires Pixels already `mvn install`ed locally and JDK 23) or consumes prebuilt `PIXELS_TRINO_CONNECTOR_ZIP` and `PIXELS_TRINO_LISTENER_ZIP` so remote Trino nodes do not need a Pixels checkout or Maven build. Installs both documented plugins by directly unzipping their zips under `plugin/` and preserving the zip's versioned top-level directory names, e.g. `pixels-trino-connector-0.2.0-SNAPSHOT` and `pixels-trino-listener-0.2.0-SNAPSHOT`; do not rename them to unversioned stable directories. Writes Trino's catalog file `etc/catalog/pixels.properties` with only catalog properties (`connector.name=pixels`, `cloud.function.switch=off`, `clean.intermediate.result=true` by default). This catalog file is **not** the same thing as Pixels' client/runtime config file `$PIXELS_HOME/etc/pixels.properties`; never copy Pixels runtime properties such as `metadata.server.host` or `etcd.hosts` into the Trino catalog. Also prepares the Trino-side lightweight Pixels client home (`TRINO_PIXELS_HOME`, default `$HOME/opt/pixels`, not `$HOME/opt/pixels/etc`) with `etc/pixels.properties` and `logs/`, then writes `export PIXELS_HOME=...` and `export PIXELS_CONFIG=$PIXELS_HOME/etc/pixels.properties` directly into the current user's bash/zsh profile (and mirrors them into `STATE_DIR/toolchain.env` for later helper-script processes). It does **not** create `~/.pixels-trino-env.sh`. Use `TRINO_PIXELS_CONFIG_SOURCE` to copy a known-good Pixels config into that client home, and use `TRINO_PIXELS_METADATA_SERVER_HOST`, `TRINO_PIXELS_TRANS_SERVER_HOST`, `TRINO_PIXELS_QUERY_SCHEDULE_SERVER_HOST`, and `TRINO_PIXELS_ETCD_HOSTS` to rewrite endpoints so Trino nodes do not fall back to `localhost`. Writes `etc/event-listener.properties` for the listener and defaults its log directory to a Trino-side path under the install parent unless `PIXELS_TRINO_LISTENER_LOG_DIR` is set. Best-effort installs `trino-cli` into `bin/trino`. +- `install_trino_shell_helpers.sh`: optional, and **always asks first**, because it bakes a specific list of remote hostnames (from `trino-deployment.env`) into a shell profile and assumes passwordless SSH between them is already set up. By default it installs only `start_trino_cluster`/`stop_trino_cluster`/`restart_trino_cluster`/`trino_cli` on the Trino coordinator recorded in `trino-deployment.env`; when invoked from another host, it copies a minimal installer over SSH and writes the coordinator user's `~/.trino-shell-helpers.sh` plus shell-profile source line there. In the generated functions, the coordinator is always operated locally and every worker over SSH. `stop_trino_cluster` stops workers before the coordinator; `restart_trino_cluster` runs stop then start. The generated start/stop/restart/client helpers explicitly export `PIXELS_HOME` and `PIXELS_CONFIG` before invoking Trino locally or over SSH, without wrapping or replacing Trino's `bin/launcher`. Set `TRINO_SHELL_HELPERS_TARGET=local` only when the user explicitly wants the functions on the current host. Leave `INSTALL_TRINO_SHELL_HELPERS` unset to be prompted `[y/N]`; set `true`/`false` to answer non-interactively once the user has actually agreed, or `ASSUME_YES=true`. + +All executable entrypoints in this skill are Bash programs. Invoke them directly +so their `#!/usr/bin/env bash` shebang is honored; do not run them with `zsh` +or `source` them into the current shell. Bash must be installed on every +deployment node even when that account's login shell is Zsh. Each entrypoint +validates the deployment account's configured login shell from passwd/NSS and +accepts only Bash or Zsh. Profile exports go to `.bashrc` or `.zshrc` +accordingly, while the generated Pixels and Trino helper files are checked in +both Bash and Zsh syntax before they are sourced. + +All of these scripts anchor paths on `$HOME` (e.g. `~/opt/...`) and detect the +current account's login shell from passwd/NSS to pick `~/.bashrc` or `~/.zshrc` before +persisting any `export`. Never assume the deployment user is named `pixels` +or `ubuntu`, and never write environment variables into a global file such +as `/etc/environment` — `scripts/lib/shell_env.sh` +centralizes this logic and is shared by `install_jdk.sh`, `install_maven.sh`, +`install_etcd.sh`, `build_install_pixels.sh`, `configure_pixels.sh`, +`check_prerequisites.sh`, `smoke_test.sh`, `install_trino.sh`, +`install_trino_cluster.sh`, `prepare_trino_cluster.sh`, +`build_pixels_trino_artifacts.sh`, +`install_trino_shell_helpers.sh`, and `progress.sh`. It also defines the +structured-result (`result_record`/`result_emit_summary`) and phase-checkpoint +(`mark_phase_done`/`phase_is_done`/`print_progress`) helpers described above. + +Each `install_*.sh` script runs as its own process, so exporting a variable +in one script does not make it visible to the next one. `docs/INSTALL.md`'s +own walkthrough handles this by running `source ~/.bashrc` right after +appending to it, but sourcing the user's real rc file from a *non*-interactive +script is unreliable (Ubuntu's default `~/.bashrc` returns immediately for +non-interactive shells, before reaching anything appended at the bottom). So +in addition to persisting into the user's real shell profile, +`shell_env.sh` also mirrors `JAVA_HOME`/`MAVEN_HOME`/`ETCD`/`PIXELS_HOME`/`PIXELS_CONFIG` +into a small `STATE_DIR/toolchain.env` file with no such guard, +and every script sources it (`load_toolchain_env`) at startup. This means +running `install_jdk.sh` then `install_maven.sh` as two separate tool calls +still works without the skill needing to manually re-source anything in +between — but if you set any of these variables yourself before running a +step, your explicit environment variable still wins. + +The skill may run simple commands from `docs/INSTALL.md` directly when that is clearer than adding another wrapper script. + +### Workflow + +The phase sequence is fixed in `skill.yaml`'s `phases` list — this section is decision guidance for each phase, not a second copy of that ordering. Before doing anything else, run `scripts/progress.sh show` to see which phases (if any) already completed in a prior session, and skip straight to the first phase that is not yet marked done instead of re-deriving state from scratch or blindly redoing finished work. After each phase actually succeeds, run `scripts/progress.sh mark ` (phase names match `skill.yaml`) so a later session — or this one, after a long gap — can pick up correctly. + +- **discover_install_target**: clarify single node vs. cluster, `PIXELS_HOME`, coordinator vs. worker roles, storage/query engine needs, whether to start services now, and whether optional components are in scope. Determine the actual login user (`whoami`) and home directory (`$HOME`) instead of assuming `pixels` or `ubuntu`; every path this skill writes is anchored on that `$HOME`. Do not proceed with defaults until the user has confirmed the topology and install locations. +- **check_prerequisites**: run `check_prerequisites.sh` and read its full structured summary before deciding what to install — see "Failure Handling" below for what to do with anything it reports as `fail`. +- **prepare_deployment_config**: run for both single-node and cluster deployments so later scripts read the same confirmed `PIXELS_HOME`, coordinator service hosts, and worker list from `deployment.env`. Prefer `prepare_deployment.sh` with no node arguments for an interactive prompt, or explicit `--coordinator ... --worker ...` arguments when the user has already provided the topology. +- **install_or_verify_jdk**: `install_jdk.sh` already skips the download when anything on the system (apt-installed JDK, a previous run, etc.) satisfies `JDK_VERSION`; it only selects and installs a Zulu build under `~/opt` when nothing qualifies. JDK 23+ is required for the documented Pixels + Trino 466 path. Ask before installing a downloaded JDK package. +- **install_or_verify_maven**: `install_maven.sh` has the same "skip if already satisfied" behavior for Maven 3.8+ (default pinned `3.9.8`, only overridden if the user asks). Ensure `mvn -v` ends up using the selected JDK. +- **install_or_verify_mysql** / **install_or_verify_etcd**: only after the user confirms credentials/ports (see Guardrails — never apply the default MySQL password silently); `install_etcd.sh` keeps localhost-only by default, asks before remote etcd access, and asks before installing its `systemd` auto-start unit. +- **build_install_pixels**: install Pixels from the repository into the confirmed `PIXELS_HOME` from `deployment.env` and place the MySQL JDBC connector under `PIXELS_HOME/lib`. In a Pixels cluster, worker nodes need this full installed `PIXELS_HOME` runtime layout to run `$PIXELS_HOME/bin/start-daemon.sh worker`, but they do **not** need to rebuild Pixels locally; build/install once, then distribute the installed `PIXELS_HOME` to worker nodes (for example with `$PIXELS_HOME/sbin/rsync-cluster.sh` after `etc/workers` is correct). +- **configure_pixels**: update `pixels.properties` only once paths, hosts, ports, database, etcd, topology, and cache settings are confirmed; it derives `pixels.var.dir` from the resolved `PIXELS_HOME`, writes `$PIXELS_HOME/etc/workers` from `PIXELS_WORKERS`, and reuses the MySQL credentials `install_mysql.sh` already confirmed. If MySQL was configured elsewhere, pass `METADATA_DB_PASSWORD` explicitly. +- **start_pixels_optional**: only when the user asks to start services or startup is part of the requested task. +- **install_shell_helpers_optional**: only if the user wants convenience functions. Install Pixels helpers on the Pixels coordinator and Trino helpers on the Trino coordinator; do not install them on the agent's current host unless that host is the corresponding coordinator or the user explicitly asks for a local helper install. Both helper scripts ask by default before writing to the target user's shell profile. +- **smoke_test**: run after configuration or startup changes; read the full structured summary (see "Failure Handling" below), don't just check the exit code. For a final deployment check, enable the relevant checks from the actual topology, e.g. `CHECK_PIXELS_LOGS=true CHECK_TRINO=true CHECK_TRINO_CLUSTER_STATUS=true CHECK_TRINO_CLI=true`. `SHOW SCHEMAS` success is enough before data is loaded; do not require returned schemas to be non-empty. +- **optional_trino_install**: only when the user explicitly asks. Clarify the topology first: coordinator vs. worker(s), how many workers, whether the coordinator also accepts scheduled work (`node-scheduler.include-coordinator`; official docs recommend `false` for a dedicated coordinator, `true` for a single-node test setup), `TRINO_INSTALL_PARENT`, `TRINO_HOME_LINK`, and `TRINO_DATA_DIR`. Never infer a fixed Trino cluster shape; use the user's topology or `prepare_trino_cluster.sh`'s confirmed output. For a Trino cluster whose remote nodes should not build Pixels, run `build_pixels_trino_artifacts.sh` once where Pixels has already been `mvn install`ed, or pass `PIXELS_TRINO_CONNECTOR_ZIP` and `PIXELS_TRINO_LISTENER_ZIP` explicitly to `install_trino_cluster.sh`; the remote nodes should only receive Trino plus those two plugins, Trino catalog/listener configs, and the lightweight Trino-side Pixels client config under `TRINO_PIXELS_HOME` (default `~/opt/pixels`). They must not be required to receive the full Pixels install, Maven, or source checkout. Run `prepare_trino_cluster.sh --setup-ssh true`, then run `install_trino_cluster.sh` **on the coordinator** to install Trino on the coordinator and fan out to every worker in parallel over SSH, instead of repeating `install_trino.sh` by hand once per node. Configure Trino's `etc/catalog/pixels.properties` as a Trino catalog file only; it is not `$PIXELS_HOME/etc/pixels.properties`. For the Trino-side Pixels client config, set `PIXELS_HOME`/`TRINO_PIXELS_HOME` to the parent directory (e.g. `~/opt/pixels`, never `~/opt/pixels/etc`) and set/export `PIXELS_CONFIG=$PIXELS_HOME/etc/pixels.properties`; all Trino nodes need this because the pixels-trino connector initializes on workers as well as the coordinator. `install_trino.sh` writes those two exports directly to the user's bash/zsh profile. When the skill itself starts, stops, checks, or runs CLI against Trino from a non-interactive SSH command, inline `PIXELS_HOME` and `PIXELS_CONFIG` in that command because remote non-interactive shells may not read `.bashrc`/`.zshrc`. Ensure `metadata.server.host`, `trans.server.host`, `query.schedule.server.host`, and `etcd.hosts` are reachable from every Trino node and are not accidental `localhost` values in a separated topology. Configure `presto.pixels.jdbc.url` only once the coordinator's endpoint and catalog/schema values are known. Offer `install_trino_shell_helpers.sh` only after the user agrees (it asks by default anyway). +- **optional_hadoop_or_monitoring**: out of scope unless the user asks — see Optional Components below. + +### Failure handling + +`check_prerequisites.sh` and `smoke_test.sh` never stop at the first failed check — they run every check and print a structured summary block (`===