diff --git a/src/main/java/com/github/ibm/mapepire/MapepireServer.java b/src/main/java/com/github/ibm/mapepire/MapepireServer.java index 82750ec..0d30f2b 100644 --- a/src/main/java/com/github/ibm/mapepire/MapepireServer.java +++ b/src/main/java/com/github/ibm/mapepire/MapepireServer.java @@ -26,6 +26,7 @@ import com.github.ibm.mapepire.authfile.AuthFile; import com.github.ibm.mapepire.certstuff.ServerCertGetter; import com.github.ibm.mapepire.certstuff.ServerCertInfo; +import com.github.ibm.mapepire.http.BlobServlet; import com.github.ibm.mapepire.http.InstallLocationServlet; import com.github.ibm.mapepire.http.Routes; import com.github.ibm.mapepire.http.VersionServlet; @@ -122,6 +123,7 @@ public static void main(final String[] _args) { context.setContextPath("/"); context.addServlet(new ServletHolder(VersionServlet.class), Routes.VERSION); context.addServlet(new ServletHolder(InstallLocationServlet.class), Routes.SOURCE); + context.addServlet(new ServletHolder(BlobServlet.class), Routes.BLOB); Constraint constraint = new Constraint(); constraint.setName("Disable TRACE"); @@ -175,8 +177,9 @@ public static void main(final String[] _args) { NativeWebSocketServletContainerInitializer.configure(context, (servletContext, nativeWebSocketConfiguration) -> { nativeWebSocketConfiguration.getPolicy().setMaxTextMessageBufferSize(65535); - // Configure max message size - int maxWsMessageSize = 50 * 1024 * 1024; // 50MB + // Max WS message size — default Integer.MAX_VALUE (unlimited). + // Can be capped via MAX_WS_MESSAGE_SIZE env var if desired. + int maxWsMessageSize = Integer.MAX_VALUE; String maxWsMessageSizeStr = System.getenv("MAX_WS_MESSAGE_SIZE"); if (StringUtils.isNonEmpty(maxWsMessageSizeStr)) { maxWsMessageSize = Integer.parseInt(maxWsMessageSizeStr); diff --git a/src/main/java/com/github/ibm/mapepire/SystemConnection.java b/src/main/java/com/github/ibm/mapepire/SystemConnection.java index e798478..fd61168 100644 --- a/src/main/java/com/github/ibm/mapepire/SystemConnection.java +++ b/src/main/java/com/github/ibm/mapepire/SystemConnection.java @@ -25,6 +25,9 @@ public enum ConnectionMethod { private final ClientSpecialRegisters m_clientRegs; private String m_applicationName; private final String clientAddress; + // Raw Base64 "user:pass" from the WebSocket Authorization header. + // Stored so BlobStore can validate HTTP /blob/{token} requests. + private String m_rawCredentials = null; public SystemConnection() throws IOException { if (!MapepireServer.isSingleMode()) { @@ -57,6 +60,16 @@ public SystemConnection(String clientHost, String clientAddress, String host, St this.m_clientRegs = new ClientSpecialRegistersRemote(clientHost, clientAddress, user); } + /** Returns the raw Base64 Authorization credentials, or {@code null} in single mode. */ + public String getRawCredentials() { + return m_rawCredentials; + } + + /** Called by {@link com.github.ibm.mapepire.ws.DbWebsocketClient} at connection time. */ + public void setRawCredentials(String rawCredentials) { + this.m_rawCredentials = rawCredentials; + } + public static boolean isRunningOnIBMi() { return System.getProperty("os.name", "").contains("400"); } diff --git a/src/main/java/com/github/ibm/mapepire/Version.java b/src/main/java/com/github/ibm/mapepire/Version.java index 225954b..3522236 100644 --- a/src/main/java/com/github/ibm/mapepire/Version.java +++ b/src/main/java/com/github/ibm/mapepire/Version.java @@ -1,5 +1,5 @@ package com.github.ibm.mapepire; public class Version { - static public final String s_compileDateTime = "2024-08-08 00:36:20 (GMT)"; - static public final String s_version = "2.0.0-rc1"; + static public final String s_compileDateTime = "2026-07-28 14:46:43 (GMT)"; + static public final String s_version = "2.3.5"; } \ No newline at end of file diff --git a/src/main/java/com/github/ibm/mapepire/http/BlobServlet.java b/src/main/java/com/github/ibm/mapepire/http/BlobServlet.java new file mode 100644 index 0000000..261fc5f --- /dev/null +++ b/src/main/java/com/github/ibm/mapepire/http/BlobServlet.java @@ -0,0 +1,106 @@ +package com.github.ibm.mapepire.http; + +import com.github.ibm.mapepire.Tracer; + +import javax.servlet.http.HttpServlet; +import javax.servlet.http.HttpServletRequest; +import javax.servlet.http.HttpServletResponse; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; + +/** + * Streams a previously-stored BLOB to the caller. + * + *

URL pattern: {@code GET /blob/{token}}

+ * + *

The caller must supply the same Basic-Auth credentials that were used + * when the originating WebSocket connection ran the query. The token itself + * is single-use and expires after the configured TTL.

+ * + *

For large BLOBs the backing data may still be spooling to disk when this + * request arrives. {@link BlobStore.BlobEntry#openStream()} will block until + * the spool is complete before any response bytes are written, ensuring the + * HTTP status code is always set correctly before the body begins.

+ */ +public class BlobServlet extends HttpServlet { + + @Override + protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws IOException { + + // ---- Extract token from path /blob/{token} ------------------------- + String pathInfo = req.getPathInfo(); + if (pathInfo == null || pathInfo.length() <= 1) { + resp.sendError(HttpServletResponse.SC_BAD_REQUEST, "Missing blob token"); + return; + } + String token = pathInfo.substring(1); // strip leading '/' + + // ---- Validate Basic Auth -------------------------------------------- + String authHeader = req.getHeader("Authorization"); + if (authHeader == null || !authHeader.startsWith("Basic ")) { + resp.setHeader("WWW-Authenticate", "Basic realm=\"mapepire\""); + resp.sendError(HttpServletResponse.SC_UNAUTHORIZED, "Authorization required"); + return; + } + String suppliedCredentials = authHeader.substring(6).trim(); // raw Base64 "user:pass" + + // ---- Consume token -------------------------------------------------- + BlobStore.BlobEntry entry = BlobStore.getInstance().consume(token); + if (entry == null) { + resp.sendError(HttpServletResponse.SC_NOT_FOUND, "Blob token not found or expired"); + return; + } + + // ---- Verify credentials match --------------------------------------- + if (!suppliedCredentials.equals(entry.credentials)) { + // Put the entry back? No — single-use, deny and discard to prevent brute-force + entry.cleanup(); + resp.setHeader("WWW-Authenticate", "Basic realm=\"mapepire\""); + resp.sendError(HttpServletResponse.SC_UNAUTHORIZED, "Invalid credentials"); + return; + } + + // ---- Wait for spool (if still in progress) and open stream ---------- + // openStream() blocks until the background spool thread finishes or the + // TTL elapses. We must open the stream BEFORE committing any response + // headers so that a spool failure can still be reported as a 500. + InputStream in = null; + try { + in = entry.openStream(); + } catch (IOException e) { + entry.cleanup(); + Tracer.err(e); + resp.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR, + "Blob data unavailable"); + return; + } + + // ---- Stream bytes --------------------------------------------------- + // Headers are committed only after openStream() succeeds, ensuring the + // status code is always meaningful. + // Use actual file size after spool completes — may differ from the JDBC-declared + // length for some IBM i LOB types. + long actualSize = entry.getActualSize(); + resp.setContentType("application/octet-stream"); + resp.setHeader("Content-Disposition", "attachment; filename=\"blob\""); + resp.setHeader("Content-Length", String.valueOf(actualSize)); + resp.setHeader("Cache-Control", "no-store"); + + try { + OutputStream out = resp.getOutputStream(); + byte[] buf = new byte[65536]; + int read; + while ((read = in.read(buf)) != -1) { + out.write(buf, 0, read); + } + out.flush(); + // Sanitise token before logging to prevent log injection. + String safeToken = token.replaceAll("[^a-fA-F0-9\\-]", "?"); + Tracer.info("BlobServlet: streamed token " + safeToken + " (" + actualSize + " bytes)"); + } finally { + try { in.close(); } catch (IOException ignored) {} + entry.cleanup(); + } + } +} diff --git a/src/main/java/com/github/ibm/mapepire/http/BlobStore.java b/src/main/java/com/github/ibm/mapepire/http/BlobStore.java new file mode 100644 index 0000000..8cf96f6 --- /dev/null +++ b/src/main/java/com/github/ibm/mapepire/http/BlobStore.java @@ -0,0 +1,391 @@ +package com.github.ibm.mapepire.http; + +import com.github.ibm.mapepire.Tracer; + +import java.io.*; +import java.nio.file.Files; +import java.sql.Blob; +import java.sql.SQLException; +import java.time.Instant; +import java.util.Iterator; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +/** + * Singleton store for BLOB tokens. + * + *

BLOBs <= {@link #MEMORY_THRESHOLD_BYTES} are held in a {@code byte[]}. + * BLOBs above that threshold are spooled to a JVM temp file in a background + * thread so that the WebSocket response (carrying the {@code blob_url}) is + * returned to the client immediately, without waiting for the full spool to + * complete. The HTTP GET on {@code /blob/{token}} will wait (up to the TTL) + * for the spool to finish before streaming bytes.

+ * + *

Each entry carries the Basic-Auth credentials of the connection that + * produced it so that {@link BlobServlet} can re-validate the caller.

+ * + *

A background thread sweeps expired entries every 30 seconds.

+ */ +public class BlobStore { + + // BLOBs larger than this are spooled to disk instead of held in memory + public static final int MEMORY_THRESHOLD_BYTES = 1024 * 1024; // 1 MB + + // Default TTL in seconds — overridable via BLOB_TOKEN_TTL env var + private static volatile long s_ttlSeconds = 60; + + private static final BlobStore s_instance = new BlobStore(); + + private final Map m_entries = new ConcurrentHashMap<>(); + private final ScheduledExecutorService m_sweeper = + Executors.newSingleThreadScheduledExecutor(r -> { + Thread t = new Thread(r, "BlobStore-sweeper"); + t.setDaemon(true); + return t; + }); + + private BlobStore() { + // Read TTL from environment on startup; override at runtime via setTtlSeconds(). + String envTtl = System.getenv("BLOB_TOKEN_TTL"); + if (envTtl != null && !envTtl.isEmpty()) { + try { + s_ttlSeconds = Long.parseLong(envTtl.trim()); + } catch (NumberFormatException e) { + Tracer.warn("Invalid BLOB_TOKEN_TTL value '" + envTtl + "', using default " + s_ttlSeconds + "s"); + } + } + m_sweeper.scheduleAtFixedRate(this::sweepExpired, 30, 30, TimeUnit.SECONDS); + } + + public static BlobStore getInstance() { + return s_instance; + } + + // ------------------------------------------------------------------------- + // TTL configuration + // ------------------------------------------------------------------------- + + public static long getTtlSeconds() { + return s_ttlSeconds; + } + + /** Override the token TTL at runtime (e.g. from a setconfig request). */ + public static void setTtlSeconds(long ttl) { + s_ttlSeconds = ttl; + } + + // ------------------------------------------------------------------------- + // Storing blobs + // ------------------------------------------------------------------------- + + /** + * Store raw bytes and return a single-use token. + * + * @param data the BLOB bytes + * @param credentials Base64-encoded "user:pass" copied from the WebSocket + * Authorization header — used to re-validate on retrieval + * @return opaque token (UUID string) to embed in the response as a URL path segment + */ + public String store(byte[] data, String credentials) throws IOException { + String token = UUID.randomUUID().toString(); + Instant expiresAt = Instant.now().plusSeconds(s_ttlSeconds); + + BlobEntry entry; + if (data.length <= MEMORY_THRESHOLD_BYTES) { + entry = BlobEntry.ofBytes(data, expiresAt, credentials); + } else { + entry = BlobEntry.ofFile(data, expiresAt, credentials); + } + m_entries.put(token, entry); + Tracer.info("BlobStore: stored token " + token + " size=" + data.length + " expires=" + expiresAt); + return token; + } + + /** + * Store a BLOB from a live JDBC {@link Blob} object and return the {@link BlobEntry} + * directly so the caller can track the async spool latch and defer closing the JDBC + * resource appropriately. + * + *

For blobs above {@link #MEMORY_THRESHOLD_BYTES} the spool to disk runs in a + * background thread so this method returns — and the caller can send the WebSocket + * response with the {@code blob_url} — immediately. The HTTP GET on + * {@code /blob/{token}} will block until the spool completes.

+ * + *

The BLOB stream is opened on the calling thread before returning so + * that the JDBC cursor can safely advance. {@link Blob#free()} is called by the spool + * thread when it finishes (large blobs) or immediately for small blobs.

+ */ + public BlobEntry storeAndReturnEntry(Blob blob, long length, String credentials) throws IOException { + String tok = UUID.randomUUID().toString(); + Instant expiresAt = Instant.now().plusSeconds(s_ttlSeconds); + + BlobEntry entry; + if (length <= MEMORY_THRESHOLD_BYTES) { + try { + byte[] bytes = readAllBytes(blob.getBinaryStream()); + blob.free(); + entry = BlobEntry.ofBytes(bytes, expiresAt, credentials); + } catch (SQLException e) { + throw new IOException("Failed to read BLOB data", e); + } + } else { + entry = BlobEntry.ofBlobAsync(blob, length, expiresAt, credentials, tok); + } + entry.m_token = tok; + m_entries.put(tok, entry); + return entry; + } + + // ------------------------------------------------------------------------- + // Retrieving blobs + // ------------------------------------------------------------------------- + + /** + * Retrieve and consume a token. Returns {@code null} if the token + * is unknown or has expired. The entry is removed immediately on retrieval + * (single-use) and any temp file is deleted after streaming. + * + *

If the backing spool is still in progress, this method returns the + * entry anyway — {@link BlobEntry#openStream()} will block until the spool + * completes or the TTL elapses.

+ */ + public BlobEntry consume(String token) { + BlobEntry entry = m_entries.remove(token); + if (entry == null) { + return null; + } + if (Instant.now().isAfter(entry.expiresAt)) { + entry.cleanup(); + return null; + } + return entry; + } + + // ------------------------------------------------------------------------- + // Java 8 compatible helpers + // ------------------------------------------------------------------------- + + private static byte[] readAllBytes(InputStream in) throws IOException { + ByteArrayOutputStream buf = new ByteArrayOutputStream(); + byte[] chunk = new byte[65536]; + int read; + while ((read = in.read(chunk)) != -1) { + buf.write(chunk, 0, read); + } + return buf.toByteArray(); + } + + // ------------------------------------------------------------------------- + // Expiry sweep + // ------------------------------------------------------------------------- + + private void sweepExpired() { + Instant now = Instant.now(); + Iterator> it = m_entries.entrySet().iterator(); + while (it.hasNext()) { + Map.Entry e = it.next(); + if (now.isAfter(e.getValue().expiresAt)) { + Tracer.info("BlobStore: expiring token " + e.getKey()); + e.getValue().cleanup(); + it.remove(); + } + } + } + + // ------------------------------------------------------------------------- + // BlobEntry — internal value type + // ------------------------------------------------------------------------- + + public static class BlobEntry { + // Exactly one of these is set once the entry is ready + private volatile byte[] m_bytes; + private volatile File m_file; + + public final long size; + // volatile so the spool thread can push expiresAt forward once done + public volatile Instant expiresAt; + public final String credentials; // Base64 "user:pass" + /** The UUID token string (without "/blob/" prefix). Set by {@link BlobStore#storeAndReturnEntry}. */ + private String m_token; + + /** Returns the UUID token string (without "/blob/" prefix). */ + public String getToken() { + return m_token; + } + + /** + * Latch counted down to zero when the backing data is fully available + * (immediately for in-memory/already-spooled entries; when the background + * spool thread finishes writing for large blobs). + */ + private final CountDownLatch m_ready; + + /** + * Non-null if the background spool thread encountered an error. + * Checked by {@link #openStream()} after the latch releases. + */ + private volatile IOException m_spoolError; + + /** True if this entry uses an async background spool thread. */ + private final boolean m_isAsyncSpool; + + private BlobEntry(byte[] bytes, File file, long size, Instant expiresAt, + String credentials, CountDownLatch ready, boolean isAsyncSpool) { + this.m_bytes = bytes; + this.m_file = file; + this.size = size; + this.expiresAt = expiresAt; + this.credentials = credentials; + this.m_ready = ready; + this.m_isAsyncSpool = isAsyncSpool; + } + + /** + * Returns true if this entry uses a background spool thread (large blob). + * False for small blobs stored synchronously in memory. + */ + public boolean isAsyncSpool() { + return m_isAsyncSpool; + } + + /** + * Block until the spool thread has finished writing all bytes to disk. + * Returns immediately for in-memory / already-complete entries. + */ + public void awaitReady() throws InterruptedException { + m_ready.await(); + } + + // -- factories -------------------------------------------------------- + + static BlobEntry ofBytes(byte[] bytes, Instant expiresAt, String credentials) { + return new BlobEntry(bytes, null, bytes.length, expiresAt, credentials, + new CountDownLatch(0), false); + } + + static BlobEntry ofFile(byte[] bytes, Instant expiresAt, String credentials) throws IOException { + File tmp = Files.createTempFile("mapepire-blob-", ".tmp").toFile(); + tmp.deleteOnExit(); + try (FileOutputStream fos = new FileOutputStream(tmp)) { + fos.write(bytes); + } + return new BlobEntry(null, tmp, bytes.length, expiresAt, credentials, + new CountDownLatch(0), false); + } + + /** + * Opens {@link Blob#getBinaryStream()} on the calling thread before + * returning, then spools the bytes to a temp file in a background thread. + * + *

The stream must be opened here — before the caller calls + * {@link java.sql.ResultSet#next()} to advance past this row — because the + * AS400 JDBC driver invalidates {@link Blob} objects (throwing + * {@code [PWS0007] Operation result set not found}) as soon as the cursor + * moves off the row that produced them, regardless of whether the + * {@link java.sql.ResultSet} is still open.

+ */ + static BlobEntry ofBlobAsync(Blob blob, long declaredLength, + Instant expiresAt, String credentials, + String tokenForLogging) throws IOException { + File tmp = Files.createTempFile("mapepire-blob-", ".tmp").toFile(); + tmp.deleteOnExit(); + + // Open the stream NOW on the calling thread, before _rs.next() is called. + final InputStream blobStream; + try { + blobStream = blob.getBinaryStream(); + } catch (SQLException e) { + try { blob.free(); } catch (SQLException ignored) {} + throw new IOException("Failed to open BLOB stream: " + e.getMessage(), e); + } + + // ready: counted down to 0 when all bytes are written to disk. + CountDownLatch ready = new CountDownLatch(1); + BlobEntry entry = new BlobEntry(null, tmp, declaredLength, + Instant.MAX /* updated by spool thread on completion */, + credentials, ready, true); + + Thread spoolThread = new Thread(() -> { + try (FileOutputStream fos = new FileOutputStream(tmp); + InputStream in = blobStream) { + byte[] buf = new byte[65536]; + int read; + while ((read = in.read(buf)) != -1) { + fos.write(buf, 0, read); + } + fos.flush(); + entry.expiresAt = Instant.now().plusSeconds(s_ttlSeconds); + } catch (IOException e) { + entry.m_spoolError = e; + entry.expiresAt = Instant.now(); + } finally { + try { blob.free(); } catch (SQLException ignored) {} + ready.countDown(); + } + }, "BlobStore-spool-" + tokenForLogging); + spoolThread.setDaemon(true); + spoolThread.start(); + + return entry; + } + + // -- accessors -------------------------------------------------------- + + /** + * Open an InputStream over this entry's data. + * + *

If the blob is still being spooled to disk, this method blocks + * until the spool finishes or the TTL elapses (whichever comes first). + * If the spool fails or the wait times out, an {@link IOException} is + * thrown and the caller should invoke {@link #cleanup()} in a + * {@code finally} block.

+ * + *

Caller is responsible for closing the returned stream.

+ */ + public InputStream openStream() throws IOException { + if (m_ready.getCount() > 0) { + // Still spooling — wait indefinitely for the JDBC read to finish. + // TTL is reset to now+TTL by the spool thread on completion so the + // client always gets a full window; no artificial timeout here. + try { + m_ready.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("Interrupted while waiting for blob spool", e); + } + } + if (m_spoolError != null) { + throw new IOException("Blob spool failed: " + m_spoolError.getMessage(), m_spoolError); + } + if (m_bytes != null) { + return new ByteArrayInputStream(m_bytes); + } + return new FileInputStream(m_file); + } + + /** + * Returns the actual number of bytes available for streaming. + * For file-backed entries this is the real file size (which may differ + * from the JDBC-declared length for some IBM i LOB types). Falls back + * to the declared size for in-memory entries. + */ + public long getActualSize() { + if (m_file != null && m_file.exists()) { + return m_file.length(); + } + return size; + } + + /** Delete the backing temp file if one exists. */ + public void cleanup() { + if (m_file != null && m_file.exists()) { + m_file.delete(); + } + } + } +} diff --git a/src/main/java/com/github/ibm/mapepire/http/Routes.java b/src/main/java/com/github/ibm/mapepire/http/Routes.java index 6420ef0..e11548d 100644 --- a/src/main/java/com/github/ibm/mapepire/http/Routes.java +++ b/src/main/java/com/github/ibm/mapepire/http/Routes.java @@ -3,4 +3,5 @@ public class Routes { public static final String VERSION = "/version"; public static final String SOURCE = "/source"; + public static final String BLOB = "/blob/*"; } diff --git a/src/main/java/com/github/ibm/mapepire/requests/BlockRetrievableRequest.java b/src/main/java/com/github/ibm/mapepire/requests/BlockRetrievableRequest.java index a16bbd6..a846ab7 100644 --- a/src/main/java/com/github/ibm/mapepire/requests/BlockRetrievableRequest.java +++ b/src/main/java/com/github/ibm/mapepire/requests/BlockRetrievableRequest.java @@ -1,6 +1,9 @@ package com.github.ibm.mapepire.requests; +import java.io.IOException; +import java.io.InputStream; import java.sql.*; +import java.util.Base64; import java.util.LinkedHashMap; import java.util.LinkedList; import java.util.List; @@ -8,7 +11,10 @@ import com.github.ibm.mapepire.ClientRequest; import com.github.ibm.mapepire.DataStreamProcessor; +import com.github.ibm.mapepire.MapepireServer; import com.github.ibm.mapepire.SystemConnection; +import com.github.ibm.mapepire.Tracer; +import com.github.ibm.mapepire.http.BlobStore; import com.google.gson.JsonObject; import com.ibm.as400.access.AS400JDBCParameterMetaData; @@ -18,6 +24,15 @@ public abstract class BlockRetrievableRequest extends ClientRequest { protected ResultSet m_rs = null; protected final boolean m_isTerseData; + /** Holds a deferred ResultSet close (with pending spool entries) until after the WS reply is sent. */ + private DataBlockFetchResult m_deferredFetchResult = null; + + /** + * Tracks large output-param BLOBs that are being async-spooled. + * Drained in processAfterReplySent() alongside result-set spools. + */ + private final List m_pendingOutputParamSpools = new LinkedList<>(); + protected BlockRetrievableRequest(DataStreamProcessor _io, SystemConnection _conn, JsonObject _reqObj) { super(_io, _conn, _reqObj); m_isTerseData = getRequestFieldBoolean("terse", false); @@ -27,8 +42,12 @@ List getNextDataBlock(final int _numRows) throws SQLException { if (m_isDone) { return new LinkedList(); } - DataBlockFetchResult result = getNextDataBlock(m_rs, _numRows, m_isTerseData); + DataBlockFetchResult result = getNextDataBlock(m_rs, _numRows, m_isTerseData, getSystemConnection()); m_isDone = result.isDone(); + // Stash deferred close info so processAfterReplySent() can close it safely. + if (result.m_deferredRs != null) { + m_deferredFetchResult = result; + } return result.m_data; } @@ -40,12 +59,13 @@ List getOutputParms(PreparedStatement _stmt) throws SQLException { CallableStatement stmt = (CallableStatement) _stmt; ParameterMetaData parmMeta = stmt.getParameterMetaData(); int numParams = parmMeta.getParameterCount(); + SystemConnection conn = getSystemConnection(); for (int i = 1; i <= numParams; ++i) { Map parmInfo = new LinkedHashMap(); parmInfo.put("index", i); parmInfo.put("type", parmMeta.getParameterTypeName(i)); parmInfo.put("precision", parmMeta.getPrecision(i)); - parmInfo.put("scale", parmMeta.getScale(numParams)); + parmInfo.put("scale", parmMeta.getScale(i)); if (parmMeta instanceof AS400JDBCParameterMetaData) { AS400JDBCParameterMetaData db2ParmMeta = (AS400JDBCParameterMetaData) parmMeta; parmInfo.put("name", db2ParmMeta.getDB2ParameterName(i)); @@ -61,6 +81,32 @@ List getOutputParms(PreparedStatement _stmt) throws SQLException { jsonValue = value.toString().trim(); } else if (value instanceof Number || value instanceof Boolean) { jsonValue = value; + } else if (value instanceof Blob) { + Blob blob = (Blob) value; + if (MapepireServer.isSingleMode()) { + try { + byte[] bytes = readAllBytes(blob.getBinaryStream()); + blob.free(); + jsonValue = Base64.getEncoder().encodeToString(bytes); + } catch (Exception e) { + Tracer.err(e); + jsonValue = null; + } + } else { + // Capture length before serializeBlob — for small blobs, + // storeAndReturnEntry calls blob.free(), so blob.length() + // after that throws HY010. + long blobLength = blob.length(); + BlobStore.BlobEntry bentry = serializeBlob(blob, blobLength, conn, + m_pendingOutputParamSpools); + jsonValue = bentry != null ? blobEntryToRef(bentry, blobLength) : null; + } + } else if (value instanceof Clob) { + Clob clob = (Clob) value; + jsonValue = clob.getSubString(1, (int) clob.length()); + clob.free(); + } else if (value instanceof byte[]) { + jsonValue = serializeBytes((byte[]) value, conn); } else { jsonValue = stmt.getString(i); } @@ -76,6 +122,13 @@ protected static class DataBlockFetchResult { private final List m_data = new LinkedList(); private boolean m_isDone = false; + /** ResultSet (and optional statement) to close after spool streams open. */ + ResultSet m_deferredRs = null; + boolean m_deferredCloseStatement = false; + + /** BlobEntries from async spools — awaited via awaitReady() before closing RS. */ + final List m_pendingSpools = new LinkedList<>(); + private DataBlockFetchResult setDone(final boolean _b) { m_isDone = _b; return this; @@ -92,10 +145,38 @@ private void add(final Object _o) { public Object getData() { return m_data; } + + /** + * Wait for all async spool threads to finish writing to disk, then close + * the deferred ResultSet/Statement. Safe to call multiple times. + * + *

We must wait for the full spool (not just stream-open) because the + * AS400 JDBC driver may close the blob's {@link InputStream} when the + * {@link ResultSet} or {@link Statement} is closed, cutting off a + * mid-read spool thread.

+ */ + void closeDeferredResultSet() { + if (m_deferredRs == null) return; + // Wait for every spool thread to finish writing before closing the RS. + for (BlobStore.BlobEntry e : m_pendingSpools) { + try { e.awaitReady(); } catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + } + } + try { + if (m_deferredCloseStatement) { + m_deferredRs.getStatement().close(); + } else { + m_deferredRs.close(); + } + } catch (SQLException ignored) {} + m_deferredRs = null; + } } protected static DataBlockFetchResult getNextDataBlock(final ResultSet _rs, final int _numRows, - final boolean _isTerseDataFormat) throws SQLException { + final boolean _isTerseDataFormat, + final SystemConnection _conn) throws SQLException { final DataBlockFetchResult ret = new DataBlockFetchResult(); if (null == _rs) { @@ -107,11 +188,22 @@ protected static DataBlockFetchResult getNextDataBlock(final ResultSet _rs, fina for (int i = 0; i < _numRows; ++i) { if (!_rs.next()) { ret.setDone(true); - Statement s = _rs.getStatement(); - if (s instanceof PreparedStatement) { - _rs.close(); + // Defer the RS close only when there are pending async spool threads that + // still need the JDBC Blob stream open. For small blobs (no async spool) + // close immediately as before to avoid HY010 "Function sequence" errors + // if the client sends the next request before processAfterReplySent() runs. + // The deferred close is resolved in processAfterReplySent() after the WS + // reply is sent, by which point all spool threads have opened their streams. + if (!ret.m_pendingSpools.isEmpty()) { + ret.m_deferredRs = _rs; + ret.m_deferredCloseStatement = !(_rs.getStatement() instanceof PreparedStatement); } else { - _rs.getStatement().close(); + // No pending spools — safe to close now + if (_rs.getStatement() instanceof PreparedStatement) { + _rs.close(); + } else { + _rs.getStatement().close(); + } } break; } @@ -132,6 +224,31 @@ protected static DataBlockFetchResult getNextDataBlock(final ResultSet _rs, fina } } else if (cellData instanceof Number || cellData instanceof Boolean) { cellDataForResponse = cellData; + } else if (cellData instanceof Blob) { + Blob blob = (Blob) cellData; + if (MapepireServer.isSingleMode()) { + // Single mode — no HTTP server, fall back to inline Base64 + try { + byte[] bytes = readAllBytes(blob.getBinaryStream()); + blob.free(); + cellDataForResponse = Base64.getEncoder().encodeToString(bytes); + } catch (Exception e) { + Tracer.err(e); + cellDataForResponse = null; + } + } else { + // Capture length before serializeBlob, which calls blob.free() + // for small blobs — calling blob.length() after free() throws HY010. + long blobLength = blob.length(); + BlobStore.BlobEntry entry = serializeBlob(blob, blobLength, _conn, ret.m_pendingSpools); + cellDataForResponse = entry != null ? blobEntryToRef(entry, blobLength) : null; + } + } else if (cellData instanceof Clob) { + Clob clob = (Clob) cellData; + cellDataForResponse = clob.getSubString(1, (int) clob.length()); + clob.free(); + } else if (cellData instanceof byte[]) { + cellDataForResponse = serializeBytes((byte[]) cellData, _conn); } else { cellDataForResponse = _rs.getString(col); } @@ -146,10 +263,93 @@ protected static DataBlockFetchResult getNextDataBlock(final ResultSet _rs, fina return ret; } + // ------------------------------------------------------------------------- + // BLOB serialization helpers + // ------------------------------------------------------------------------- + + /** + * Daemon-mode only: stores blob in {@link BlobStore} and tracks any async spool + * entry in the supplied list so the caller can defer closing the JDBC resource. + * + *

Accepts either a {@link DataBlockFetchResult#m_pendingSpools} list (result-set + * path) or {@link #m_pendingOutputParamSpools} (output-parameter path).

+ */ + private static BlobStore.BlobEntry serializeBlob(Blob blob, long length, + SystemConnection conn, + List pendingSpools) { + try { + // BlobStore.storeAndReturnEntry owns blob.free() from this point on + BlobStore.BlobEntry entry = BlobStore.getInstance().storeAndReturnEntry(blob, length, conn.getRawCredentials()); + // Only track entries that have a live async spool thread — small blobs + // are stored synchronously and their latch is already at 0. + if (entry != null && entry.isAsyncSpool()) { + pendingSpools.add(entry); + } + return entry; + } catch (IOException e) { + Tracer.err(e); + try { blob.free(); } catch (Exception ignored) {} + return null; + } + } + + private static Map blobEntryToRef(BlobStore.BlobEntry entry, long length) { + Map ref = new LinkedHashMap<>(); + ref.put("blob_url", "/blob/" + entry.getToken()); + ref.put("size", length); + return ref; + } + + private static Object serializeBytes(byte[] bytes, SystemConnection conn) { + if (MapepireServer.isSingleMode()) { + return Base64.getEncoder().encodeToString(bytes); + } + try { + String token = BlobStore.getInstance().store(bytes, conn.getRawCredentials()); + Map ref = new LinkedHashMap<>(); + ref.put("blob_url", "/blob/" + token); + ref.put("size", (long) bytes.length); + return ref; + } catch (IOException e) { + Tracer.err(e); + return null; + } + } + + private static byte[] readAllBytes(InputStream in) throws IOException { + java.io.ByteArrayOutputStream buf = new java.io.ByteArrayOutputStream(); + byte[] chunk = new byte[65536]; + int read; + while ((read = in.read(chunk)) != -1) { + buf.write(chunk, 0, read); + } + return buf.toByteArray(); + } + public boolean isDone() { return m_isDone; } + /** + * Called by {@link com.github.ibm.mapepire.ClientRequest#run()} after the WebSocket + * reply has been sent. Closes any ResultSet that was deferred to allow async blob + * spool threads to finish writing, and waits for any output-parameter async spools. + */ + @Override + protected void processAfterReplySent() { + if (m_deferredFetchResult != null) { + m_deferredFetchResult.closeDeferredResultSet(); + m_deferredFetchResult = null; + } + // Drain any async output-parameter blob spools. + for (BlobStore.BlobEntry e : m_pendingOutputParamSpools) { + try { e.awaitReady(); } catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + } + } + m_pendingOutputParamSpools.clear(); + } + protected Map getResultMetaDataForResponse() throws SQLException { return getResultMetaDataForResponse(this.m_rs.getMetaData(), getSystemConnection()); } @@ -173,6 +373,12 @@ public static Map getResultMetaDataForResponse(ResultSetMetaData columnAttrs.put("readOnly", _md.isReadOnly(i)); columnAttrs.put("writeable", _md.isWritable(i)); columnAttrs.put("table", _md.getTableName(i)); + // Signal to clients that this column's value will be a blob_url object + // in daemon mode rather than an inline string value. + int colType = _md.getColumnType(i); + boolean isBlobType = colType == Types.BLOB || colType == Types.BINARY + || colType == Types.VARBINARY || colType == Types.LONGVARBINARY; + columnAttrs.put("blob_as_url", isBlobType && !MapepireServer.isSingleMode()); columnMetaData.add(columnAttrs); } metaData.put("columns", columnMetaData); diff --git a/src/main/java/com/github/ibm/mapepire/requests/DoVe.java b/src/main/java/com/github/ibm/mapepire/requests/DoVe.java index 1f601d9..ccb876b 100644 --- a/src/main/java/com/github/ibm/mapepire/requests/DoVe.java +++ b/src/main/java/com/github/ibm/mapepire/requests/DoVe.java @@ -114,7 +114,7 @@ public void go() throws Exception { } ResultSet veData = callStmt.getResultSet(); addReplyData("vemetadata", getResultMetaDataForResponse(veData.getMetaData(), getSystemConnection())); - addReplyData("vedata", super.getNextDataBlock(veData, Integer.MAX_VALUE, m_isTerseData).getData()); + addReplyData("vedata", super.getNextDataBlock(veData, Integer.MAX_VALUE, m_isTerseData, getSystemConnection()).getData()); if (null != this.m_rs) { addReplyData("metadata", getResultMetaDataForResponse()); addReplyData("data", super.getNextDataBlock(numRows)); diff --git a/src/main/java/com/github/ibm/mapepire/requests/PrepareSql.java b/src/main/java/com/github/ibm/mapepire/requests/PrepareSql.java index 6087772..3682c11 100644 --- a/src/main/java/com/github/ibm/mapepire/requests/PrepareSql.java +++ b/src/main/java/com/github/ibm/mapepire/requests/PrepareSql.java @@ -117,4 +117,19 @@ public boolean isDone() { return true; } + /** + * Delegate to the execute task so that any deferred ResultSet close + * (needed when a large-blob async spool is in progress) is triggered + * after the WebSocket reply is sent. Without this, m_deferredFetchResult + * is set on the PreparedExecute instance but processAfterReplySent() would + * only run on this PrepareSql instance, silently skipping the deferred close. + */ + @Override + protected void processAfterReplySent() { + super.processAfterReplySent(); + if (m_executeTask != null) { + m_executeTask.processAfterReplySent(); + } + } + } diff --git a/src/main/java/com/github/ibm/mapepire/requests/PreparedExecute.java b/src/main/java/com/github/ibm/mapepire/requests/PreparedExecute.java index 0f7a6a4..3b03ef3 100644 --- a/src/main/java/com/github/ibm/mapepire/requests/PreparedExecute.java +++ b/src/main/java/com/github/ibm/mapepire/requests/PreparedExecute.java @@ -6,9 +6,11 @@ import java.sql.SQLException; import java.sql.Types; import java.util.Arrays; +import java.util.Base64; import java.util.LinkedList; import com.github.ibm.mapepire.DataStreamProcessor; +import com.github.ibm.mapepire.http.BlobStore; import com.google.gson.JsonArray; import com.google.gson.JsonElement; import com.google.gson.JsonObject; @@ -78,12 +80,42 @@ private void addJsonArrayParameters(PreparedStatement stmt, JsonArray arr) throw } else if (stmt instanceof CallableStatement && ParameterMetaData.parameterModeInOut == stmt.getParameterMetaData().getParameterMode(i)) { ((CallableStatement) stmt).registerOutParameter(i, stmt.getParameterMetaData().getParameterType(i)); - stmt.setString(i, element.getAsString()); + setParameterValue(stmt, i, element.getAsString()); } else { - stmt.setString(i, element.getAsString()); + setParameterValue(stmt, i, element.getAsString()); } } } } + private static void setParameterValue(PreparedStatement stmt, int i, String value) throws SQLException { + int paramType = stmt.getParameterMetaData().getParameterType(i); + switch (paramType) { + case Types.BLOB: + case Types.BINARY: + case Types.VARBINARY: + case Types.LONGVARBINARY: { + byte[] bytes = Base64.getDecoder().decode(value); + if (bytes.length > BlobStore.MEMORY_THRESHOLD_BYTES) { + // Stream large blobs directly to avoid double-buffering in heap + stmt.setBinaryStream(i, new java.io.ByteArrayInputStream(bytes), bytes.length); + } else { + stmt.setBytes(i, bytes); + } + break; + } + // Character LOBs — pass through as a plain string. Explicit cases here + // make intent clear and protect the default branch from future changes. + case Types.CLOB: + case Types.NCLOB: + case Types.LONGVARCHAR: + case Types.LONGNVARCHAR: + stmt.setString(i, value); + break; + default: + stmt.setString(i, value); + break; + } + } + } diff --git a/src/main/java/com/github/ibm/mapepire/requests/RunSqlMore.java b/src/main/java/com/github/ibm/mapepire/requests/RunSqlMore.java index 26eff77..2dc9ce6 100644 --- a/src/main/java/com/github/ibm/mapepire/requests/RunSqlMore.java +++ b/src/main/java/com/github/ibm/mapepire/requests/RunSqlMore.java @@ -20,4 +20,14 @@ protected void go() throws Exception { addReplyData("is_done", m_prev.isDone()); } + /** + * Delegate to the previous request so that any deferred ResultSet close + * or pending output-param async spool is resolved after the WebSocket reply + * is sent — even though it is this RunSqlMore instance whose run() executes. (Fix 4) + */ + @Override + protected void processAfterReplySent() { + m_prev.processAfterReplySent(); + } + } diff --git a/src/main/java/com/github/ibm/mapepire/ws/DbSocketCreator.java b/src/main/java/com/github/ibm/mapepire/ws/DbSocketCreator.java index 20a5fb5..5bfdfa1 100644 --- a/src/main/java/com/github/ibm/mapepire/ws/DbSocketCreator.java +++ b/src/main/java/com/github/ibm/mapepire/ws/DbSocketCreator.java @@ -97,7 +97,7 @@ public Object createWebSocket(ServletUpgradeRequest jettyServerUpgradeRequest, S password[i] = (char) decoded[colonIndex + 1 + i]; } - return new DbWebsocketClient(jettyServerUpgradeRequest.getRemoteHostName(), jettyServerUpgradeRequest.getRemoteAddress(), DbSocketCreator.getHost(), username, password); + return new DbWebsocketClient(jettyServerUpgradeRequest.getRemoteHostName(), jettyServerUpgradeRequest.getRemoteAddress(), DbSocketCreator.getHost(), username, password, asBase64); } catch (IOException e) { // TODO Auto-generated catch block e.printStackTrace(); diff --git a/src/main/java/com/github/ibm/mapepire/ws/DbWebsocketClient.java b/src/main/java/com/github/ibm/mapepire/ws/DbWebsocketClient.java index abcaf45..028ed2f 100644 --- a/src/main/java/com/github/ibm/mapepire/ws/DbWebsocketClient.java +++ b/src/main/java/com/github/ibm/mapepire/ws/DbWebsocketClient.java @@ -14,9 +14,10 @@ public class DbWebsocketClient extends WebSocketAdapter { private final CountDownLatch closureLatch = new CountDownLatch(1); private final DataStreamProcessor io; - DbWebsocketClient(String clientHost, String clientAddress, String host, String user, char[] pass) throws IOException { + DbWebsocketClient(String clientHost, String clientAddress, String host, String user, char[] pass, String rawCredentials) throws IOException { super(); SystemConnection conn = new SystemConnection(clientHost, clientAddress, host, user, pass); + conn.setRawCredentials(rawCredentials); io = getDataStream(this, conn); }