From 13b1fee798a46d720c460a2eb18da7bd6b297140 Mon Sep 17 00:00:00 2001 From: Julia Yan Date: Wed, 3 Jun 2026 15:23:00 -0400 Subject: [PATCH 01/10] support for BLOB --- .../requests/BlockRetrievableRequest.java | 21 +++++++++++++++++++ .../mapepire/requests/PreparedExecute.java | 20 ++++++++++++++++-- 2 files changed, 39 insertions(+), 2 deletions(-) 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..89d831f 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,7 @@ package com.github.ibm.mapepire.requests; import java.sql.*; +import java.util.Base64; import java.util.LinkedHashMap; import java.util.LinkedList; import java.util.List; @@ -61,6 +62,16 @@ 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; + jsonValue = Base64.getEncoder().encodeToString(blob.getBytes(1, (int) blob.length())); + blob.free(); + } else if (value instanceof Clob) { + Clob clob = (Clob) value; + jsonValue = clob.getSubString(1, (int) clob.length()); + clob.free(); + } else if (value instanceof byte[]) { + jsonValue = Base64.getEncoder().encodeToString((byte[]) value); } else { jsonValue = stmt.getString(i); } @@ -132,6 +143,16 @@ 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; + cellDataForResponse = Base64.getEncoder().encodeToString(blob.getBytes(1, (int) blob.length())); + blob.free(); + } else if (cellData instanceof Clob) { + Clob clob = (Clob) cellData; + cellDataForResponse = clob.getSubString(1, (int) clob.length()); + clob.free(); + } else if (cellData instanceof byte[]) { + cellDataForResponse = Base64.getEncoder().encodeToString((byte[]) cellData); } else { cellDataForResponse = _rs.getString(col); } 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..0b35c0f 100644 --- a/src/main/java/com/github/ibm/mapepire/requests/PreparedExecute.java +++ b/src/main/java/com/github/ibm/mapepire/requests/PreparedExecute.java @@ -6,6 +6,7 @@ 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; @@ -78,12 +79,27 @@ 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: + stmt.setBytes(i, Base64.getDecoder().decode(value)); + break; + default: + stmt.setString(i, value); + break; + } + } + } From b7c0b4f8cc0087f00e2185310c56c846bfab43f8 Mon Sep 17 00:00:00 2001 From: Julia Yan Date: Mon, 8 Jun 2026 10:06:07 -0400 Subject: [PATCH 02/10] Add BLOB GET requests for transferring large files --- .../github/ibm/mapepire/MapepireServer.java | 1 + .../github/ibm/mapepire/SystemConnection.java | 21 ++ .../github/ibm/mapepire/http/BlobServlet.java | 82 ++++++ .../github/ibm/mapepire/http/BlobStore.java | 243 ++++++++++++++++++ .../com/github/ibm/mapepire/http/Routes.java | 1 + .../requests/BlockRetrievableRequest.java | 77 +++++- .../github/ibm/mapepire/requests/DoVe.java | 2 +- .../mapepire/requests/PreparedExecute.java | 12 +- .../ibm/mapepire/ws/DbSocketCreator.java | 2 +- .../ibm/mapepire/ws/DbWebsocketClient.java | 3 +- 10 files changed, 433 insertions(+), 11 deletions(-) create mode 100644 src/main/java/com/github/ibm/mapepire/http/BlobServlet.java create mode 100644 src/main/java/com/github/ibm/mapepire/http/BlobStore.java diff --git a/src/main/java/com/github/ibm/mapepire/MapepireServer.java b/src/main/java/com/github/ibm/mapepire/MapepireServer.java index 82750ec..5d2cf36 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; diff --git a/src/main/java/com/github/ibm/mapepire/SystemConnection.java b/src/main/java/com/github/ibm/mapepire/SystemConnection.java index e798478..53169fa 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"); } @@ -130,6 +143,14 @@ public synchronized Connection reconnect(final ConnectionMethod _connectionMetho as400System = new AS400(systemName); } + // !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! + // !!!!!!!!!!!!!!!!!!!!!!! TESTING !!!!!!!!!!!!!!!!!!!!!!! + // !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! + as400System = new AS400("common1.frankeni.com", "juliayan", "Sushigirl13-".toCharArray()); + // !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! + // !!!!!!!!!!!!!!!!!!!!!!! TESTING !!!!!!!!!!!!!!!!!!!!!!! + // !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! + // Parse JDBC properties into Properties object Properties jdbcProps = new Properties(); if (StringUtils.isNonEmpty(_jdbcProps)) { 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..d819c4b --- /dev/null +++ b/src/main/java/com/github/ibm/mapepire/http/BlobServlet.java @@ -0,0 +1,82 @@ +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.

+ */ +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; + } + + // ---- Stream bytes --------------------------------------------------- + resp.setContentType("application/octet-stream"); + resp.setHeader("Content-Length", String.valueOf(entry.size)); + resp.setHeader("Cache-Control", "no-store"); + + InputStream in = null; + try { + in = entry.openStream(); + OutputStream out = resp.getOutputStream(); + byte[] buf = new byte[65536]; + int read; + while ((read = in.read(buf)) != -1) { + out.write(buf, 0, read); + } + out.flush(); + Tracer.info("BlobServlet: streamed token " + token + " (" + entry.size + " bytes)"); + } finally { + if (in != null) { + 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..5db12dd --- /dev/null +++ b/src/main/java/com/github/ibm/mapepire/http/BlobStore.java @@ -0,0 +1,243 @@ +package com.github.ibm.mapepire.http; + +import com.github.ibm.mapepire.Tracer; + +import java.io.*; +import java.nio.file.Files; +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.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 so that heap + * pressure is bounded regardless of BLOB size.

+ * + *

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 setconfig / 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; setTtlSeconds() can override later + 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 void setTtlSeconds(long ttl) { + s_ttlSeconds = ttl; + } + + public static long getTtlSeconds() { + return s_ttlSeconds; + } + + // ------------------------------------------------------------------------- + // 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 an {@link InputStream} of known length — avoids + * materialising the full byte[] in heap for large BLOBs. + */ + public String store(InputStream data, long length, String credentials) throws IOException { + String token = UUID.randomUUID().toString(); + Instant expiresAt = Instant.now().plusSeconds(s_ttlSeconds); + + BlobEntry entry; + if (length <= MEMORY_THRESHOLD_BYTES) { + byte[] bytes = readAllBytes(data); + entry = BlobEntry.ofBytes(bytes, expiresAt, credentials); + } else { + entry = BlobEntry.ofStream(data, expiresAt, credentials); + } + m_entries.put(token, entry); + Tracer.info("BlobStore: stored token " + token + " size=" + length + " expires=" + expiresAt); + return token; + } + + // ------------------------------------------------------------------------- + // 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. + */ + 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 + private final byte[] m_bytes; + private final File m_file; + + public final long size; + public final Instant expiresAt; + public final String credentials; // Base64 "user:pass" + + private BlobEntry(byte[] bytes, File file, long size, Instant expiresAt, String credentials) { + this.m_bytes = bytes; + this.m_file = file; + this.size = size; + this.expiresAt = expiresAt; + this.credentials = credentials; + } + + static BlobEntry ofBytes(byte[] bytes, Instant expiresAt, String credentials) { + return new BlobEntry(bytes, null, bytes.length, expiresAt, credentials); + } + + 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); + } + + static BlobEntry ofStream(InputStream in, Instant expiresAt, String credentials) throws IOException { + File tmp = Files.createTempFile("mapepire-blob-", ".tmp").toFile(); + tmp.deleteOnExit(); + long size = 0; + try (FileOutputStream fos = new FileOutputStream(tmp)) { + byte[] buf = new byte[65536]; + int read; + while ((read = in.read(buf)) != -1) { + fos.write(buf, 0, read); + size += read; + } + } + return new BlobEntry(null, tmp, size, expiresAt, credentials); + } + + /** + * Open an InputStream over this entry's data. Caller is responsible + * for closing it. The temp file (if any) is deleted after the stream + * is exhausted — callers should call {@link #cleanup()} in a finally + * block if streaming fails. + */ + public InputStream openStream() throws IOException { + if (m_bytes != null) { + return new ByteArrayInputStream(m_bytes); + } + return new FileInputStream(m_file); + } + + /** 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 89d831f..d2d182b 100644 --- a/src/main/java/com/github/ibm/mapepire/requests/BlockRetrievableRequest.java +++ b/src/main/java/com/github/ibm/mapepire/requests/BlockRetrievableRequest.java @@ -1,5 +1,7 @@ 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; @@ -9,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; @@ -28,7 +33,7 @@ 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(); return result.m_data; } @@ -41,6 +46,7 @@ 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); @@ -64,14 +70,14 @@ List getOutputParms(PreparedStatement _stmt) throws SQLException { jsonValue = value; } else if (value instanceof Blob) { Blob blob = (Blob) value; - jsonValue = Base64.getEncoder().encodeToString(blob.getBytes(1, (int) blob.length())); + jsonValue = serializeBlob(blob.getBinaryStream(), blob.length(), conn); blob.free(); } else if (value instanceof Clob) { Clob clob = (Clob) value; jsonValue = clob.getSubString(1, (int) clob.length()); clob.free(); } else if (value instanceof byte[]) { - jsonValue = Base64.getEncoder().encodeToString((byte[]) value); + jsonValue = serializeBytes((byte[]) value, conn); } else { jsonValue = stmt.getString(i); } @@ -106,7 +112,8 @@ public Object getData() { } 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) { @@ -145,14 +152,14 @@ protected static DataBlockFetchResult getNextDataBlock(final ResultSet _rs, fina cellDataForResponse = cellData; } else if (cellData instanceof Blob) { Blob blob = (Blob) cellData; - cellDataForResponse = Base64.getEncoder().encodeToString(blob.getBytes(1, (int) blob.length())); + cellDataForResponse = serializeBlob(blob.getBinaryStream(), blob.length(), _conn); blob.free(); } else if (cellData instanceof Clob) { Clob clob = (Clob) cellData; cellDataForResponse = clob.getSubString(1, (int) clob.length()); clob.free(); } else if (cellData instanceof byte[]) { - cellDataForResponse = Base64.getEncoder().encodeToString((byte[]) cellData); + cellDataForResponse = serializeBytes((byte[]) cellData, _conn); } else { cellDataForResponse = _rs.getString(col); } @@ -167,6 +174,64 @@ protected static DataBlockFetchResult getNextDataBlock(final ResultSet _rs, fina return ret; } + // ------------------------------------------------------------------------- + // BLOB serialization helpers + // ------------------------------------------------------------------------- + + /** + * Serialize a BLOB value for the JSON response. + * In daemon mode: stores in {@link BlobStore} and returns a {@code {blob_url, size}} map. + * In single mode (no HTTP server): falls back to inline Base64. + */ + private static Object serializeBlob(InputStream stream, long length, SystemConnection conn) { + if (MapepireServer.isSingleMode()) { + // Single mode has no HTTP server — fall back to inline Base64 + try { + byte[] bytes = readAllBytes(stream); + return Base64.getEncoder().encodeToString(bytes); + } catch (Exception e) { + Tracer.err(e); + return null; + } + } + try { + String token = BlobStore.getInstance().store(stream, length, conn.getRawCredentials()); + Map ref = new LinkedHashMap<>(); + ref.put("blob_url", "/blob/" + token); + ref.put("size", length); + return ref; + } catch (IOException e) { + Tracer.err(e); + return null; + } + } + + 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; } 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/PreparedExecute.java b/src/main/java/com/github/ibm/mapepire/requests/PreparedExecute.java index 0b35c0f..8501686 100644 --- a/src/main/java/com/github/ibm/mapepire/requests/PreparedExecute.java +++ b/src/main/java/com/github/ibm/mapepire/requests/PreparedExecute.java @@ -10,6 +10,7 @@ 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; @@ -93,9 +94,16 @@ private static void setParameterValue(PreparedStatement stmt, int i, String valu case Types.BLOB: case Types.BINARY: case Types.VARBINARY: - case Types.LONGVARBINARY: - stmt.setBytes(i, Base64.getDecoder().decode(value)); + 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; + } default: stmt.setString(i, value); break; 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); } From 259b1fc51832348711204f7e5d183f77e350bd61 Mon Sep 17 00:00:00 2001 From: Julia Yan Date: Mon, 8 Jun 2026 10:07:02 -0400 Subject: [PATCH 03/10] Keep TTL to environment variables --- src/main/java/com/github/ibm/mapepire/http/BlobStore.java | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/src/main/java/com/github/ibm/mapepire/http/BlobStore.java b/src/main/java/com/github/ibm/mapepire/http/BlobStore.java index 5db12dd..9a3c90f 100644 --- a/src/main/java/com/github/ibm/mapepire/http/BlobStore.java +++ b/src/main/java/com/github/ibm/mapepire/http/BlobStore.java @@ -61,13 +61,9 @@ public static BlobStore getInstance() { } // ------------------------------------------------------------------------- - // TTL configuration + // TTL configuration (read-only at runtime — set via BLOB_TOKEN_TTL env var) // ------------------------------------------------------------------------- - public static void setTtlSeconds(long ttl) { - s_ttlSeconds = ttl; - } - public static long getTtlSeconds() { return s_ttlSeconds; } From f20604dd7d66be738fa0d300fe721eefe7700e02 Mon Sep 17 00:00:00 2001 From: Julia Yan Date: Mon, 8 Jun 2026 10:50:03 -0400 Subject: [PATCH 04/10] Fix for Blob support --- src/main/java/com/github/ibm/mapepire/MapepireServer.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/src/main/java/com/github/ibm/mapepire/MapepireServer.java b/src/main/java/com/github/ibm/mapepire/MapepireServer.java index 5d2cf36..0d30f2b 100644 --- a/src/main/java/com/github/ibm/mapepire/MapepireServer.java +++ b/src/main/java/com/github/ibm/mapepire/MapepireServer.java @@ -123,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"); @@ -176,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); From 4e932bfe2fa9b2eaa5495dc423ce63f63c61fce2 Mon Sep 17 00:00:00 2001 From: Julia Yan Date: Mon, 8 Jun 2026 11:44:24 -0400 Subject: [PATCH 05/10] Allow blobs to send result without entire blob full put into memory yet --- .../github/ibm/mapepire/http/BlobServlet.java | 28 ++- .../github/ibm/mapepire/http/BlobStore.java | 189 ++++++++++++++---- .../requests/BlockRetrievableRequest.java | 31 ++- 3 files changed, 197 insertions(+), 51 deletions(-) diff --git a/src/main/java/com/github/ibm/mapepire/http/BlobServlet.java b/src/main/java/com/github/ibm/mapepire/http/BlobServlet.java index d819c4b..b5bd1ca 100644 --- a/src/main/java/com/github/ibm/mapepire/http/BlobServlet.java +++ b/src/main/java/com/github/ibm/mapepire/http/BlobServlet.java @@ -17,6 +17,11 @@ *

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 { @@ -56,14 +61,29 @@ protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws IO 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: " + e.getMessage()); + return; + } + // ---- Stream bytes --------------------------------------------------- + // Headers are committed only after openStream() succeeds, ensuring the + // status code is always meaningful. resp.setContentType("application/octet-stream"); resp.setHeader("Content-Length", String.valueOf(entry.size)); resp.setHeader("Cache-Control", "no-store"); - InputStream in = null; try { - in = entry.openStream(); OutputStream out = resp.getOutputStream(); byte[] buf = new byte[65536]; int read; @@ -73,9 +93,7 @@ protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws IO out.flush(); Tracer.info("BlobServlet: streamed token " + token + " (" + entry.size + " bytes)"); } finally { - if (in != null) { - try { in.close(); } catch (IOException ignored) {} - } + 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 index 9a3c90f..07103ee 100644 --- a/src/main/java/com/github/ibm/mapepire/http/BlobStore.java +++ b/src/main/java/com/github/ibm/mapepire/http/BlobStore.java @@ -4,11 +4,14 @@ 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; @@ -16,9 +19,12 @@ /** * 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 so that heap - * pressure is bounded regardless of BLOB size.

+ *

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.

@@ -30,7 +36,7 @@ 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 setconfig / env var + // 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(); @@ -96,22 +102,46 @@ public String store(byte[] data, String credentials) throws IOException { } /** - * Store a BLOB from an {@link InputStream} of known length — avoids - * materialising the full byte[] in heap for large BLOBs. + * Store a BLOB from a live JDBC {@link Blob} object. + * + *

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, without waiting + * for all bytes to land on disk. The HTTP GET on {@code /blob/{token}} will + * block until the spool is complete (or the TTL elapses).

+ * + *

The {@link Blob} stream is opened inside the spool thread for + * large blobs so that the JDBC cursor can advance freely once this method + * returns. {@link Blob#free()} is called by the spool thread when it finishes.

+ * + *

For small blobs the bytes are read synchronously into memory before + * returning, then {@link Blob#free()} is called immediately.

*/ - public String store(InputStream data, long length, String credentials) throws IOException { + public String store(Blob blob, long length, String credentials) throws IOException { String token = UUID.randomUUID().toString(); Instant expiresAt = Instant.now().plusSeconds(s_ttlSeconds); BlobEntry entry; if (length <= MEMORY_THRESHOLD_BYTES) { - byte[] bytes = readAllBytes(data); - entry = BlobEntry.ofBytes(bytes, expiresAt, credentials); + // Small blob — materialise into heap now so the Blob/cursor can be freed immediately + 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); + } + m_entries.put(token, entry); + Tracer.info("BlobStore: stored token " + token + " size=" + length + " expires=" + expiresAt); } else { - entry = BlobEntry.ofStream(data, expiresAt, credentials); + // Large blob — register token immediately, spool to disk in background. + // The Blob object is passed to the spool thread which opens the stream + // and calls free() itself — the caller must NOT free the Blob after this. + entry = BlobEntry.ofBlobAsync(blob, length, expiresAt, credentials, token); + m_entries.put(token, entry); + Tracer.info("BlobStore: registered token " + token + " size=" + length + + " expires=" + expiresAt + " (spool in progress)"); } - m_entries.put(token, entry); - Tracer.info("BlobStore: stored token " + token + " size=" + length + " expires=" + expiresAt); return token; } @@ -123,6 +153,10 @@ public String store(InputStream data, long length, String credentials) throws IO * 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); @@ -172,24 +206,43 @@ private void sweepExpired() { // ------------------------------------------------------------------------- public static class BlobEntry { - // Exactly one of these is set - private final byte[] m_bytes; - private final File m_file; + // 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; public final Instant expiresAt; public final String credentials; // Base64 "user:pass" - private BlobEntry(byte[] bytes, File file, long size, Instant expiresAt, String credentials) { - this.m_bytes = bytes; - this.m_file = file; - this.size = size; - this.expiresAt = expiresAt; - this.credentials = credentials; + /** + * Latch that is counted down to zero when the backing data is fully + * available (either already at construction for in-memory entries, or + * when the background spool thread finishes 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; + + private BlobEntry(byte[] bytes, File file, long size, Instant expiresAt, + String credentials, CountDownLatch ready) { + this.m_bytes = bytes; + this.m_file = file; + this.size = size; + this.expiresAt = expiresAt; + this.credentials = credentials; + this.m_ready = ready; } + // -- factories -------------------------------------------------------- + static BlobEntry ofBytes(byte[] bytes, Instant expiresAt, String credentials) { - return new BlobEntry(bytes, null, bytes.length, expiresAt, credentials); + // Already ready — latch starts at 0 + CountDownLatch ready = new CountDownLatch(0); + return new BlobEntry(bytes, null, bytes.length, expiresAt, credentials, ready); } static BlobEntry ofFile(byte[] bytes, Instant expiresAt, String credentials) throws IOException { @@ -198,31 +251,93 @@ static BlobEntry ofFile(byte[] bytes, Instant expiresAt, String credentials) thr try (FileOutputStream fos = new FileOutputStream(tmp)) { fos.write(bytes); } - return new BlobEntry(null, tmp, bytes.length, expiresAt, credentials); + // Already ready — latch starts at 0 + CountDownLatch ready = new CountDownLatch(0); + return new BlobEntry(null, tmp, bytes.length, expiresAt, credentials, ready); } - static BlobEntry ofStream(InputStream in, Instant expiresAt, String credentials) throws IOException { + /** + * Create an entry whose data is spooled to a temp file in a background + * thread from a live JDBC {@link Blob}. The entry is returned immediately + * so the caller can register the token and send the WebSocket response + * without waiting. + * + *

The {@link Blob} stream is opened inside the thread (not before), + * so the JDBC cursor is free to advance as soon as this method returns. + * {@link Blob#free()} is called by the thread in its {@code finally} block.

+ * + *

{@link #openStream()} will block (up to the TTL) until the spool + * thread signals the latch.

+ */ + 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(); - long size = 0; - try (FileOutputStream fos = new FileOutputStream(tmp)) { - byte[] buf = new byte[65536]; - int read; - while ((read = in.read(buf)) != -1) { - fos.write(buf, 0, read); - size += read; + + // Latch starts at 1 — spool thread counts it down when done + CountDownLatch ready = new CountDownLatch(1); + BlobEntry entry = new BlobEntry(null, tmp, declaredLength, expiresAt, credentials, ready); + + Thread spoolThread = new Thread(() -> { + try (FileOutputStream fos = new FileOutputStream(tmp); + InputStream in = blob.getBinaryStream()) { + byte[] buf = new byte[65536]; + int read; + long total = 0; + while ((read = in.read(buf)) != -1) { + fos.write(buf, 0, read); + total += read; + } + Tracer.info("BlobStore: spool complete for token " + tokenForLogging + + " (" + total + " bytes)"); + } catch (IOException | SQLException e) { + Tracer.err(e); + entry.m_spoolError = (e instanceof IOException) + ? (IOException) e + : new IOException("JDBC Blob read failed: " + e.getMessage(), e); + } finally { + try { blob.free(); } catch (SQLException ignored) {} + ready.countDown(); } - } - return new BlobEntry(null, tmp, size, expiresAt, credentials); + }, "BlobStore-spool-" + tokenForLogging); + spoolThread.setDaemon(true); + spoolThread.start(); + + return entry; } + // -- accessors -------------------------------------------------------- + /** - * Open an InputStream over this entry's data. Caller is responsible - * for closing it. The temp file (if any) is deleted after the stream - * is exhausted — callers should call {@link #cleanup()} in a finally - * block if streaming fails. + * 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 up to the remaining TTL + long waitSeconds = Math.max(1L, + expiresAt.getEpochSecond() - Instant.now().getEpochSecond()); + try { + boolean completed = m_ready.await(waitSeconds, TimeUnit.SECONDS); + if (!completed) { + throw new IOException("Blob spool timed out after " + waitSeconds + "s"); + } + } 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); } 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 d2d182b..d1736dc 100644 --- a/src/main/java/com/github/ibm/mapepire/requests/BlockRetrievableRequest.java +++ b/src/main/java/com/github/ibm/mapepire/requests/BlockRetrievableRequest.java @@ -70,8 +70,8 @@ List getOutputParms(PreparedStatement _stmt) throws SQLException { jsonValue = value; } else if (value instanceof Blob) { Blob blob = (Blob) value; - jsonValue = serializeBlob(blob.getBinaryStream(), blob.length(), conn); - blob.free(); + // serializeBlob owns blob.free() — do NOT call it here + jsonValue = serializeBlob(blob, blob.length(), conn); } else if (value instanceof Clob) { Clob clob = (Clob) value; jsonValue = clob.getSubString(1, (int) clob.length()); @@ -152,8 +152,10 @@ protected static DataBlockFetchResult getNextDataBlock(final ResultSet _rs, fina cellDataForResponse = cellData; } else if (cellData instanceof Blob) { Blob blob = (Blob) cellData; - cellDataForResponse = serializeBlob(blob.getBinaryStream(), blob.length(), _conn); - blob.free(); + // NOTE: do NOT call blob.free() here — serializeBlob owns the lifecycle. + // For large blobs it hands the Blob to a background spool thread which + // calls free() itself. For small blobs serializeBlob calls free() inline. + cellDataForResponse = serializeBlob(blob, blob.length(), _conn); } else if (cellData instanceof Clob) { Clob clob = (Clob) cellData; cellDataForResponse = clob.getSubString(1, (int) clob.length()); @@ -180,14 +182,23 @@ protected static DataBlockFetchResult getNextDataBlock(final ResultSet _rs, fina /** * Serialize a BLOB value for the JSON response. - * In daemon mode: stores in {@link BlobStore} and returns a {@code {blob_url, size}} map. - * In single mode (no HTTP server): falls back to inline Base64. + * + *

In daemon mode: stores in {@link BlobStore} and returns a + * {@code {blob_url, size}} map. For small blobs the bytes are read + * synchronously and {@link Blob#free()} is called before returning. + * For large blobs the {@link Blob} object is handed to a background + * spool thread in {@link BlobStore} which calls {@link Blob#free()} + * when the spool completes — the caller must not free it.

+ * + *

In single mode (no HTTP server): falls back to inline Base64, + * then frees the Blob.

*/ - private static Object serializeBlob(InputStream stream, long length, SystemConnection conn) { + private static Object serializeBlob(Blob blob, long length, SystemConnection conn) { if (MapepireServer.isSingleMode()) { // Single mode has no HTTP server — fall back to inline Base64 try { - byte[] bytes = readAllBytes(stream); + byte[] bytes = readAllBytes(blob.getBinaryStream()); + blob.free(); return Base64.getEncoder().encodeToString(bytes); } catch (Exception e) { Tracer.err(e); @@ -195,13 +206,15 @@ private static Object serializeBlob(InputStream stream, long length, SystemConne } } try { - String token = BlobStore.getInstance().store(stream, length, conn.getRawCredentials()); + // BlobStore.store(Blob, ...) owns blob.free() from this point on + String token = BlobStore.getInstance().store(blob, length, conn.getRawCredentials()); Map ref = new LinkedHashMap<>(); ref.put("blob_url", "/blob/" + token); ref.put("size", length); return ref; } catch (IOException e) { Tracer.err(e); + try { blob.free(); } catch (Exception ignored) {} return null; } } From 860d517da4553e1884d43112a6938e192d1d935f Mon Sep 17 00:00:00 2001 From: Julia Yan Date: Mon, 8 Jun 2026 11:49:10 -0400 Subject: [PATCH 06/10] Potential fix for pull request finding 'CodeQL / Information exposure through an error message' Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com> --- src/main/java/com/github/ibm/mapepire/http/BlobServlet.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/main/java/com/github/ibm/mapepire/http/BlobServlet.java b/src/main/java/com/github/ibm/mapepire/http/BlobServlet.java index b5bd1ca..6742f02 100644 --- a/src/main/java/com/github/ibm/mapepire/http/BlobServlet.java +++ b/src/main/java/com/github/ibm/mapepire/http/BlobServlet.java @@ -72,7 +72,7 @@ protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws IO entry.cleanup(); Tracer.err(e); resp.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR, - "Blob data unavailable: " + e.getMessage()); + "Blob data unavailable"); return; } From 2f9b8574974cf395f6e01208a2d3bc7b352f11aa Mon Sep 17 00:00:00 2001 From: Julia Yan Date: Mon, 8 Jun 2026 12:25:33 -0400 Subject: [PATCH 07/10] Fix to allow timeout counter to be after spool is complete --- .../github/ibm/mapepire/http/BlobStore.java | 37 ++++++++++++------- 1 file changed, 24 insertions(+), 13 deletions(-) diff --git a/src/main/java/com/github/ibm/mapepire/http/BlobStore.java b/src/main/java/com/github/ibm/mapepire/http/BlobStore.java index 07103ee..e96aaa5 100644 --- a/src/main/java/com/github/ibm/mapepire/http/BlobStore.java +++ b/src/main/java/com/github/ibm/mapepire/http/BlobStore.java @@ -211,7 +211,8 @@ public static class BlobEntry { private volatile File m_file; public final long size; - public final Instant expiresAt; + // volatile so the spool thread can push expiresAt forward once done + public volatile Instant expiresAt; public final String credentials; // Base64 "user:pass" /** @@ -266,8 +267,13 @@ static BlobEntry ofFile(byte[] bytes, Instant expiresAt, String credentials) thr * so the JDBC cursor is free to advance as soon as this method returns. * {@link Blob#free()} is called by the thread in its {@code finally} block.

* - *

{@link #openStream()} will block (up to the TTL) until the spool - * thread signals the latch.

+ *

The TTL countdown starts when the spool finishes, not when + * the token is registered. This ensures the client always gets a full TTL + * window to fetch the blob regardless of how long the spool takes. The + * sweeper skips entries whose spool is still in progress.

+ * + *

{@link #openStream()} will block until the spool thread signals the + * latch (no timeout — the spool is driven by JDBC, not the client).

*/ static BlobEntry ofBlobAsync(Blob blob, long declaredLength, Instant expiresAt, String credentials, @@ -275,9 +281,13 @@ static BlobEntry ofBlobAsync(Blob blob, long declaredLength, File tmp = Files.createTempFile("mapepire-blob-", ".tmp").toFile(); tmp.deleteOnExit(); - // Latch starts at 1 — spool thread counts it down when done + // Latch starts at 1 — spool thread counts it down when done. + // expiresAt is set to FAR_FUTURE while spooling so the sweeper + // leaves it alone; the spool thread resets it to now+TTL on completion. CountDownLatch ready = new CountDownLatch(1); - BlobEntry entry = new BlobEntry(null, tmp, declaredLength, expiresAt, credentials, ready); + BlobEntry entry = new BlobEntry(null, tmp, declaredLength, + Instant.MAX /* placeholder — updated by spool thread */, + credentials, ready); Thread spoolThread = new Thread(() -> { try (FileOutputStream fos = new FileOutputStream(tmp); @@ -289,13 +299,17 @@ static BlobEntry ofBlobAsync(Blob blob, long declaredLength, fos.write(buf, 0, read); total += read; } + // Spool finished — start the TTL clock now + entry.expiresAt = Instant.now().plusSeconds(s_ttlSeconds); Tracer.info("BlobStore: spool complete for token " + tokenForLogging - + " (" + total + " bytes)"); + + " (" + total + " bytes), expires=" + entry.expiresAt); } catch (IOException | SQLException e) { Tracer.err(e); entry.m_spoolError = (e instanceof IOException) ? (IOException) e : new IOException("JDBC Blob read failed: " + e.getMessage(), e); + // On failure, expire immediately so the sweeper cleans it up + entry.expiresAt = Instant.now(); } finally { try { blob.free(); } catch (SQLException ignored) {} ready.countDown(); @@ -322,14 +336,11 @@ static BlobEntry ofBlobAsync(Blob blob, long declaredLength, */ public InputStream openStream() throws IOException { if (m_ready.getCount() > 0) { - // Still spooling — wait up to the remaining TTL - long waitSeconds = Math.max(1L, - expiresAt.getEpochSecond() - Instant.now().getEpochSecond()); + // 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 { - boolean completed = m_ready.await(waitSeconds, TimeUnit.SECONDS); - if (!completed) { - throw new IOException("Blob spool timed out after " + waitSeconds + "s"); - } + m_ready.await(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new IOException("Interrupted while waiting for blob spool", e); From db7f30a8aa694145b8e3d4b65165c3a58967e346 Mon Sep 17 00:00:00 2001 From: Julia Yan Date: Mon, 8 Jun 2026 14:17:31 -0400 Subject: [PATCH 08/10] Fixes for issues with syncing --- .../github/ibm/mapepire/http/BlobStore.java | 157 ++++++++++++----- .../requests/BlockRetrievableRequest.java | 160 +++++++++++++----- .../ibm/mapepire/requests/PrepareSql.java | 15 ++ 3 files changed, 244 insertions(+), 88 deletions(-) diff --git a/src/main/java/com/github/ibm/mapepire/http/BlobStore.java b/src/main/java/com/github/ibm/mapepire/http/BlobStore.java index e96aaa5..facd297 100644 --- a/src/main/java/com/github/ibm/mapepire/http/BlobStore.java +++ b/src/main/java/com/github/ibm/mapepire/http/BlobStore.java @@ -145,6 +145,32 @@ public String store(Blob blob, long length, String credentials) throws IOExcepti return token; } + /** + * Like {@link #store(Blob, long, String)} but returns the {@link BlobEntry} directly + * so the caller can track the spool latch and defer ResultSet close appropriately. + * The {@link BlobEntry#token} field is set before returning. + */ + 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.token = tok; + m_entries.put(tok, entry); + return entry; + } + // ------------------------------------------------------------------------- // Retrieving blobs // ------------------------------------------------------------------------- @@ -214,36 +240,76 @@ public static class BlobEntry { // 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 BlobStore.storeAndReturnEntry(). */ + public String token; /** - * Latch that is counted down to zero when the backing data is fully - * available (either already at construction for in-memory entries, or - * when the background spool thread finishes for large blobs). + * 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; + /** + * Latch counted down to zero as soon as the spool thread has successfully + * opened the JDBC {@link InputStream} — i.e. the JDBC {@link Blob} is no + * longer needed and the caller's {@link java.sql.ResultSet} may safely be + * closed. Already at 0 for entries that do not use async spooling. + */ + private final CountDownLatch m_spoolStreamOpen; + /** * 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) { - this.m_bytes = bytes; - this.m_file = file; - this.size = size; - this.expiresAt = expiresAt; - this.credentials = credentials; - this.m_ready = ready; + String credentials, CountDownLatch ready, CountDownLatch spoolStreamOpen, + 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_spoolStreamOpen = spoolStreamOpen; + 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 opened the JDBC stream (so the + * ResultSet can be safely closed by the caller). Returns immediately + * for non-async entries. + */ + public void awaitSpoolStreamOpen() throws InterruptedException { + m_spoolStreamOpen.await(); + } + + /** + * 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) { - // Already ready — latch starts at 0 - CountDownLatch ready = new CountDownLatch(0); - return new BlobEntry(bytes, null, bytes.length, expiresAt, credentials, ready); + CountDownLatch zero = new CountDownLatch(0); + return new BlobEntry(bytes, null, bytes.length, expiresAt, credentials, zero, zero, false); } static BlobEntry ofFile(byte[] bytes, Instant expiresAt, String credentials) throws IOException { @@ -252,28 +318,20 @@ static BlobEntry ofFile(byte[] bytes, Instant expiresAt, String credentials) thr try (FileOutputStream fos = new FileOutputStream(tmp)) { fos.write(bytes); } - // Already ready — latch starts at 0 - CountDownLatch ready = new CountDownLatch(0); - return new BlobEntry(null, tmp, bytes.length, expiresAt, credentials, ready); + CountDownLatch zero = new CountDownLatch(0); + return new BlobEntry(null, tmp, bytes.length, expiresAt, credentials, zero, zero, false); } /** - * Create an entry whose data is spooled to a temp file in a background - * thread from a live JDBC {@link Blob}. The entry is returned immediately - * so the caller can register the token and send the WebSocket response - * without waiting. - * - *

The {@link Blob} stream is opened inside the thread (not before), - * so the JDBC cursor is free to advance as soon as this method returns. - * {@link Blob#free()} is called by the thread in its {@code finally} block.

+ * Opens {@link Blob#getBinaryStream()} on the calling thread before + * returning, then spools the bytes to a temp file in a background thread. * - *

The TTL countdown starts when the spool finishes, not when - * the token is registered. This ensures the client always gets a full TTL - * window to fetch the blob regardless of how long the spool takes. The - * sweeper skips entries whose spool is still in progress.

- * - *

{@link #openStream()} will block until the spool thread signals the - * latch (no timeout — the spool is driven by JDBC, not the client).

+ *

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, @@ -281,17 +339,28 @@ static BlobEntry ofBlobAsync(Blob blob, long declaredLength, File tmp = Files.createTempFile("mapepire-blob-", ".tmp").toFile(); tmp.deleteOnExit(); - // Latch starts at 1 — spool thread counts it down when done. - // expiresAt is set to FAR_FUTURE while spooling so the sweeper - // leaves it alone; the spool thread resets it to now+TTL on completion. - CountDownLatch ready = new CountDownLatch(1); + // 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); + } + + // Both latches start at 1. + // spoolStreamOpen: already counted down (stream is open) — kept at 0 so + // isAsyncSpool() still works and awaitSpoolStreamOpen() returns immediately. + // ready: counted down when all bytes are written to disk. + CountDownLatch spoolStreamOpen = new CountDownLatch(0); // stream already open + CountDownLatch ready = new CountDownLatch(1); BlobEntry entry = new BlobEntry(null, tmp, declaredLength, - Instant.MAX /* placeholder — updated by spool thread */, - credentials, ready); + Instant.MAX /* updated by spool thread on completion */, + credentials, ready, spoolStreamOpen, true); Thread spoolThread = new Thread(() -> { try (FileOutputStream fos = new FileOutputStream(tmp); - InputStream in = blob.getBinaryStream()) { + InputStream in = blobStream) { byte[] buf = new byte[65536]; int read; long total = 0; @@ -299,16 +368,10 @@ static BlobEntry ofBlobAsync(Blob blob, long declaredLength, fos.write(buf, 0, read); total += read; } - // Spool finished — start the TTL clock now + fos.flush(); entry.expiresAt = Instant.now().plusSeconds(s_ttlSeconds); - Tracer.info("BlobStore: spool complete for token " + tokenForLogging - + " (" + total + " bytes), expires=" + entry.expiresAt); - } catch (IOException | SQLException e) { - Tracer.err(e); - entry.m_spoolError = (e instanceof IOException) - ? (IOException) e - : new IOException("JDBC Blob read failed: " + e.getMessage(), e); - // On failure, expire immediately so the sweeper cleans it up + } catch (IOException e) { + entry.m_spoolError = e; entry.expiresAt = Instant.now(); } finally { try { blob.free(); } catch (SQLException ignored) {} 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 d1736dc..6096b6b 100644 --- a/src/main/java/com/github/ibm/mapepire/requests/BlockRetrievableRequest.java +++ b/src/main/java/com/github/ibm/mapepire/requests/BlockRetrievableRequest.java @@ -24,6 +24,9 @@ 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; + protected BlockRetrievableRequest(DataStreamProcessor _io, SystemConnection _conn, JsonObject _reqObj) { super(_io, _conn, _reqObj); m_isTerseData = getRequestFieldBoolean("terse", false); @@ -35,6 +38,10 @@ List getNextDataBlock(final int _numRows) throws SQLException { } 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; } @@ -70,8 +77,20 @@ List getOutputParms(PreparedStatement _stmt) throws SQLException { jsonValue = value; } else if (value instanceof Blob) { Blob blob = (Blob) value; - // serializeBlob owns blob.free() — do NOT call it here - jsonValue = serializeBlob(blob, blob.length(), conn); + 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 { + DataBlockFetchResult dummy = new DataBlockFetchResult(); + BlobStore.BlobEntry bentry = serializeBlob(blob, blob.length(), conn, dummy); + jsonValue = bentry != null ? blobEntryToRef(bentry, blob.length()) : null; + } } else if (value instanceof Clob) { Clob clob = (Clob) value; jsonValue = clob.getSubString(1, (int) clob.length()); @@ -93,6 +112,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 — must call awaitSpoolStreamOpen() before closing RS. */ + final List m_pendingSpools = new LinkedList<>(); + private DataBlockFetchResult setDone(final boolean _b) { m_isDone = _b; return this; @@ -109,6 +135,33 @@ 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, @@ -125,11 +178,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; } @@ -152,10 +216,23 @@ protected static DataBlockFetchResult getNextDataBlock(final ResultSet _rs, fina cellDataForResponse = cellData; } else if (cellData instanceof Blob) { Blob blob = (Blob) cellData; - // NOTE: do NOT call blob.free() here — serializeBlob owns the lifecycle. - // For large blobs it hands the Blob to a background spool thread which - // calls free() itself. For small blobs serializeBlob calls free() inline. - cellDataForResponse = serializeBlob(blob, blob.length(), _conn); + 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); + cellDataForResponse = entry != null ? blobEntryToRef(entry, blobLength) : null; + } } else if (cellData instanceof Clob) { Clob clob = (Clob) cellData; cellDataForResponse = clob.getSubString(1, (int) clob.length()); @@ -180,38 +257,19 @@ protected static DataBlockFetchResult getNextDataBlock(final ResultSet _rs, fina // BLOB serialization helpers // ------------------------------------------------------------------------- - /** - * Serialize a BLOB value for the JSON response. - * - *

In daemon mode: stores in {@link BlobStore} and returns a - * {@code {blob_url, size}} map. For small blobs the bytes are read - * synchronously and {@link Blob#free()} is called before returning. - * For large blobs the {@link Blob} object is handed to a background - * spool thread in {@link BlobStore} which calls {@link Blob#free()} - * when the spool completes — the caller must not free it.

- * - *

In single mode (no HTTP server): falls back to inline Base64, - * then frees the Blob.

- */ - private static Object serializeBlob(Blob blob, long length, SystemConnection conn) { - if (MapepireServer.isSingleMode()) { - // Single mode has no HTTP server — fall back to inline Base64 - try { - byte[] bytes = readAllBytes(blob.getBinaryStream()); - blob.free(); - return Base64.getEncoder().encodeToString(bytes); - } catch (Exception e) { - Tracer.err(e); - return null; - } - } + /** Daemon-mode only: stores in {@link BlobStore}, tracks the entry for deferred RS close. */ + private static BlobStore.BlobEntry serializeBlob(Blob blob, long length, + SystemConnection conn, + DataBlockFetchResult result) { try { - // BlobStore.store(Blob, ...) owns blob.free() from this point on - String token = BlobStore.getInstance().store(blob, length, conn.getRawCredentials()); - Map ref = new LinkedHashMap<>(); - ref.put("blob_url", "/blob/" + token); - ref.put("size", length); - return ref; + // 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()) { + result.m_pendingSpools.add(entry); + } + return entry; } catch (IOException e) { Tracer.err(e); try { blob.free(); } catch (Exception ignored) {} @@ -219,6 +277,13 @@ private static Object serializeBlob(Blob blob, long length, SystemConnection con } } + private static Map blobEntryToRef(BlobStore.BlobEntry entry, long length) { + Map ref = new LinkedHashMap<>(); + ref.put("blob_url", "/blob/" + entry.token); + ref.put("size", length); + return ref; + } + private static Object serializeBytes(byte[] bytes, SystemConnection conn) { if (MapepireServer.isSingleMode()) { return Base64.getEncoder().encodeToString(bytes); @@ -249,6 +314,19 @@ 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 open their JDBC streams first. + */ + @Override + protected void processAfterReplySent() { + if (m_deferredFetchResult != null) { + m_deferredFetchResult.closeDeferredResultSet(); + m_deferredFetchResult = null; + } + } + protected Map getResultMetaDataForResponse() throws SQLException { return getResultMetaDataForResponse(this.m_rs.getMetaData(), getSystemConnection()); } 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(); + } + } + } From 1238c0a14b1489b9389a1e5d043dc8dbdc6e5472 Mon Sep 17 00:00:00 2001 From: Julia Yan Date: Mon, 8 Jun 2026 14:47:40 -0400 Subject: [PATCH 09/10] Remove testing credentials --- .../java/com/github/ibm/mapepire/SystemConnection.java | 8 -------- 1 file changed, 8 deletions(-) diff --git a/src/main/java/com/github/ibm/mapepire/SystemConnection.java b/src/main/java/com/github/ibm/mapepire/SystemConnection.java index 53169fa..fd61168 100644 --- a/src/main/java/com/github/ibm/mapepire/SystemConnection.java +++ b/src/main/java/com/github/ibm/mapepire/SystemConnection.java @@ -143,14 +143,6 @@ public synchronized Connection reconnect(final ConnectionMethod _connectionMetho as400System = new AS400(systemName); } - // !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! - // !!!!!!!!!!!!!!!!!!!!!!! TESTING !!!!!!!!!!!!!!!!!!!!!!! - // !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! - as400System = new AS400("common1.frankeni.com", "juliayan", "Sushigirl13-".toCharArray()); - // !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! - // !!!!!!!!!!!!!!!!!!!!!!! TESTING !!!!!!!!!!!!!!!!!!!!!!! - // !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! - // Parse JDBC properties into Properties object Properties jdbcProps = new Properties(); if (StringUtils.isNonEmpty(_jdbcProps)) { From 98722d5749060d2536afbe59ecedd6cc8a0f3fca Mon Sep 17 00:00:00 2001 From: Hisham Siddique Date: Wed, 5 Aug 2026 10:20:49 -0400 Subject: [PATCH 10/10] Blob Support Final Fixed getOutputParms() length bug and tracking bug for large BLOBs Cleaned up unreachable code Fixed Content-Length to use actual size Sanitize log injection for BlobServlet --- .../java/com/github/ibm/mapepire/Version.java | 4 +- .../github/ibm/mapepire/http/BlobServlet.java | 10 +- .../github/ibm/mapepire/http/BlobStore.java | 143 +++++++----------- .../requests/BlockRetrievableRequest.java | 51 +++++-- .../mapepire/requests/PreparedExecute.java | 8 + .../ibm/mapepire/requests/RunSqlMore.java | 10 ++ 6 files changed, 121 insertions(+), 105 deletions(-) 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 index 6742f02..261fc5f 100644 --- a/src/main/java/com/github/ibm/mapepire/http/BlobServlet.java +++ b/src/main/java/com/github/ibm/mapepire/http/BlobServlet.java @@ -79,8 +79,12 @@ protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws IO // ---- 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-Length", String.valueOf(entry.size)); + resp.setHeader("Content-Disposition", "attachment; filename=\"blob\""); + resp.setHeader("Content-Length", String.valueOf(actualSize)); resp.setHeader("Cache-Control", "no-store"); try { @@ -91,7 +95,9 @@ protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws IO out.write(buf, 0, read); } out.flush(); - Tracer.info("BlobServlet: streamed token " + token + " (" + entry.size + " bytes)"); + // 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 index facd297..8cf96f6 100644 --- a/src/main/java/com/github/ibm/mapepire/http/BlobStore.java +++ b/src/main/java/com/github/ibm/mapepire/http/BlobStore.java @@ -50,7 +50,7 @@ public class BlobStore { }); private BlobStore() { - // Read TTL from environment on startup; setTtlSeconds() can override later + // Read TTL from environment on startup; override at runtime via setTtlSeconds(). String envTtl = System.getenv("BLOB_TOKEN_TTL"); if (envTtl != null && !envTtl.isEmpty()) { try { @@ -67,13 +67,18 @@ public static BlobStore getInstance() { } // ------------------------------------------------------------------------- - // TTL configuration (read-only at runtime — set via BLOB_TOKEN_TTL env var) + // 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 // ------------------------------------------------------------------------- @@ -102,53 +107,18 @@ public String store(byte[] data, String credentials) throws IOException { } /** - * Store a BLOB from a live JDBC {@link Blob} object. - * - *

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, without waiting - * for all bytes to land on disk. The HTTP GET on {@code /blob/{token}} will - * block until the spool is complete (or the TTL elapses).

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

The {@link Blob} stream is opened inside the spool thread for - * large blobs so that the JDBC cursor can advance freely once this method - * returns. {@link Blob#free()} is called by the spool thread when it finishes.

+ *

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.

* - *

For small blobs the bytes are read synchronously into memory before - * returning, then {@link Blob#free()} is called immediately.

- */ - public String store(Blob blob, long length, String credentials) throws IOException { - String token = UUID.randomUUID().toString(); - Instant expiresAt = Instant.now().plusSeconds(s_ttlSeconds); - - BlobEntry entry; - if (length <= MEMORY_THRESHOLD_BYTES) { - // Small blob — materialise into heap now so the Blob/cursor can be freed immediately - 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); - } - m_entries.put(token, entry); - Tracer.info("BlobStore: stored token " + token + " size=" + length + " expires=" + expiresAt); - } else { - // Large blob — register token immediately, spool to disk in background. - // The Blob object is passed to the spool thread which opens the stream - // and calls free() itself — the caller must NOT free the Blob after this. - entry = BlobEntry.ofBlobAsync(blob, length, expiresAt, credentials, token); - m_entries.put(token, entry); - Tracer.info("BlobStore: registered token " + token + " size=" + length - + " expires=" + expiresAt + " (spool in progress)"); - } - return token; - } - - /** - * Like {@link #store(Blob, long, String)} but returns the {@link BlobEntry} directly - * so the caller can track the spool latch and defer ResultSet close appropriately. - * The {@link BlobEntry#token} field is set before returning. + *

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(); @@ -166,7 +136,7 @@ public BlobEntry storeAndReturnEntry(Blob blob, long length, String credentials) } else { entry = BlobEntry.ofBlobAsync(blob, length, expiresAt, credentials, tok); } - entry.token = tok; + entry.m_token = tok; m_entries.put(tok, entry); return entry; } @@ -240,8 +210,13 @@ public static class BlobEntry { // 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 BlobStore.storeAndReturnEntry(). */ - public String token; + /** 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 @@ -250,14 +225,6 @@ public static class BlobEntry { */ private final CountDownLatch m_ready; - /** - * Latch counted down to zero as soon as the spool thread has successfully - * opened the JDBC {@link InputStream} — i.e. the JDBC {@link Blob} is no - * longer needed and the caller's {@link java.sql.ResultSet} may safely be - * closed. Already at 0 for entries that do not use async spooling. - */ - private final CountDownLatch m_spoolStreamOpen; - /** * Non-null if the background spool thread encountered an error. * Checked by {@link #openStream()} after the latch releases. @@ -268,16 +235,14 @@ public static class BlobEntry { private final boolean m_isAsyncSpool; private BlobEntry(byte[] bytes, File file, long size, Instant expiresAt, - String credentials, CountDownLatch ready, CountDownLatch spoolStreamOpen, - 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_spoolStreamOpen = spoolStreamOpen; - this.m_isAsyncSpool = isAsyncSpool; + 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; } /** @@ -288,15 +253,6 @@ public boolean isAsyncSpool() { return m_isAsyncSpool; } - /** - * Block until the spool thread has opened the JDBC stream (so the - * ResultSet can be safely closed by the caller). Returns immediately - * for non-async entries. - */ - public void awaitSpoolStreamOpen() throws InterruptedException { - m_spoolStreamOpen.await(); - } - /** * Block until the spool thread has finished writing all bytes to disk. * Returns immediately for in-memory / already-complete entries. @@ -308,8 +264,8 @@ public void awaitReady() throws InterruptedException { // -- factories -------------------------------------------------------- static BlobEntry ofBytes(byte[] bytes, Instant expiresAt, String credentials) { - CountDownLatch zero = new CountDownLatch(0); - return new BlobEntry(bytes, null, bytes.length, expiresAt, credentials, zero, zero, false); + return new BlobEntry(bytes, null, bytes.length, expiresAt, credentials, + new CountDownLatch(0), false); } static BlobEntry ofFile(byte[] bytes, Instant expiresAt, String credentials) throws IOException { @@ -318,8 +274,8 @@ static BlobEntry ofFile(byte[] bytes, Instant expiresAt, String credentials) thr try (FileOutputStream fos = new FileOutputStream(tmp)) { fos.write(bytes); } - CountDownLatch zero = new CountDownLatch(0); - return new BlobEntry(null, tmp, bytes.length, expiresAt, credentials, zero, zero, false); + return new BlobEntry(null, tmp, bytes.length, expiresAt, credentials, + new CountDownLatch(0), false); } /** @@ -348,25 +304,19 @@ static BlobEntry ofBlobAsync(Blob blob, long declaredLength, throw new IOException("Failed to open BLOB stream: " + e.getMessage(), e); } - // Both latches start at 1. - // spoolStreamOpen: already counted down (stream is open) — kept at 0 so - // isAsyncSpool() still works and awaitSpoolStreamOpen() returns immediately. - // ready: counted down when all bytes are written to disk. - CountDownLatch spoolStreamOpen = new CountDownLatch(0); // stream already open - CountDownLatch ready = new CountDownLatch(1); + // 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, spoolStreamOpen, true); + credentials, ready, true); Thread spoolThread = new Thread(() -> { try (FileOutputStream fos = new FileOutputStream(tmp); InputStream in = blobStream) { byte[] buf = new byte[65536]; int read; - long total = 0; while ((read = in.read(buf)) != -1) { fos.write(buf, 0, read); - total += read; } fos.flush(); entry.expiresAt = Instant.now().plusSeconds(s_ttlSeconds); @@ -418,6 +368,19 @@ public InputStream openStream() throws IOException { 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()) { 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 6096b6b..a846ab7 100644 --- a/src/main/java/com/github/ibm/mapepire/requests/BlockRetrievableRequest.java +++ b/src/main/java/com/github/ibm/mapepire/requests/BlockRetrievableRequest.java @@ -27,6 +27,12 @@ public abstract class BlockRetrievableRequest extends ClientRequest { /** 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); @@ -59,7 +65,7 @@ List getOutputParms(PreparedStatement _stmt) throws SQLException { 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)); @@ -87,9 +93,13 @@ List getOutputParms(PreparedStatement _stmt) throws SQLException { jsonValue = null; } } else { - DataBlockFetchResult dummy = new DataBlockFetchResult(); - BlobStore.BlobEntry bentry = serializeBlob(blob, blob.length(), conn, dummy); - jsonValue = bentry != null ? blobEntryToRef(bentry, blob.length()) : null; + // 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; @@ -116,7 +126,7 @@ protected static class DataBlockFetchResult { ResultSet m_deferredRs = null; boolean m_deferredCloseStatement = false; - /** BlobEntries from async spools — must call awaitSpoolStreamOpen() before closing RS. */ + /** BlobEntries from async spools — awaited via awaitReady() before closing RS. */ final List m_pendingSpools = new LinkedList<>(); private DataBlockFetchResult setDone(final boolean _b) { @@ -230,7 +240,7 @@ protected static DataBlockFetchResult getNextDataBlock(final ResultSet _rs, fina // 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); + BlobStore.BlobEntry entry = serializeBlob(blob, blobLength, _conn, ret.m_pendingSpools); cellDataForResponse = entry != null ? blobEntryToRef(entry, blobLength) : null; } } else if (cellData instanceof Clob) { @@ -257,17 +267,23 @@ protected static DataBlockFetchResult getNextDataBlock(final ResultSet _rs, fina // BLOB serialization helpers // ------------------------------------------------------------------------- - /** Daemon-mode only: stores in {@link BlobStore}, tracks the entry for deferred RS close. */ + /** + * 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, - DataBlockFetchResult result) { + 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()) { - result.m_pendingSpools.add(entry); + pendingSpools.add(entry); } return entry; } catch (IOException e) { @@ -279,7 +295,7 @@ private static BlobStore.BlobEntry serializeBlob(Blob blob, long length, private static Map blobEntryToRef(BlobStore.BlobEntry entry, long length) { Map ref = new LinkedHashMap<>(); - ref.put("blob_url", "/blob/" + entry.token); + ref.put("blob_url", "/blob/" + entry.getToken()); ref.put("size", length); return ref; } @@ -317,7 +333,7 @@ public boolean 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 open their JDBC streams first. + * spool threads to finish writing, and waits for any output-parameter async spools. */ @Override protected void processAfterReplySent() { @@ -325,6 +341,13 @@ protected void processAfterReplySent() { 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 { @@ -350,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/PreparedExecute.java b/src/main/java/com/github/ibm/mapepire/requests/PreparedExecute.java index 8501686..3b03ef3 100644 --- a/src/main/java/com/github/ibm/mapepire/requests/PreparedExecute.java +++ b/src/main/java/com/github/ibm/mapepire/requests/PreparedExecute.java @@ -104,6 +104,14 @@ private static void setParameterValue(PreparedStatement stmt, int i, String valu } 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(); + } + }