protocol: Query Table refresh tokens and EndStreamAction

server/opensharing-server-core/src/main/java/io/opensharing/asset/table/DeltaSharingCapabilities.java

@@ -45,9 +45,19 @@ public static ResponseFormat choose(String header) {
 
   /** Value for the response {@code delta-sharing-capabilities} header: the format actually used. */
   public static String responded(String requestHeader) {
-    return choose(requestHeader) == ResponseFormat.DELTA
-        ? "responseformat=delta"
-        : "responseformat=parquet";
+    return responded(requestHeader, false);
+  }
+
+  public static String responded(String requestHeader, boolean includeEndStreamAction) {
+    String format =
+        choose(requestHeader) == ResponseFormat.DELTA
+            ? "responseformat=delta"
+            : "responseformat=parquet";
+    return includeEndStreamAction ? format + ";includeendstreamaction=true" : format;
+  }
+
+  public static boolean includeEndStreamAction(String header) {
+    return "true".equals(parse(header).get("includeendstreamaction"));
   }
 
   /**

server/opensharing-server-core/src/main/java/io/opensharing/asset/table/ProtocolTables.java

@@ -157,7 +157,8 @@ public TemporaryTableCredentialsResponse temporaryCredentials(
   /**
    * One snapshot of a table in the share that supports url access, as of the latest version,
    * {@code version}, or an ISO-8601 {@code timestamp} in {@code request}, with presigned file
-   * URLs. Pushdown hints and startingVersion queries are not implemented.
+   * URLs. A latest-snapshot query can ask for a refresh token, and a later query can pass it back
+   * to re-sign the same version. Pushdown hints and startingVersion queries are not implemented.
    */
   public DeltaTableQueryReader.Result query(
       RecipientPrincipal recipient,
@@ -180,12 +181,24 @@ public DeltaTableQueryReader.Result query(
     if (hasPushdownHint(body)) {
       throw ApiException.notImplemented("predicate and limit pushdown is not supported");
     }
+    boolean historical = body.version() != null || body.timestamp() != null;
+    boolean includeRefreshToken = Boolean.TRUE.equals(body.includeRefreshToken());
+    if (includeRefreshToken && historical) {
+      throw ApiException.invalidParameter(
+          "includeRefreshToken cannot be used when querying a specific version.");
+    }
+    if (body.refreshToken() != null && historical) {
+      throw ApiException.invalidParameter(
+          "refreshToken cannot be used when querying a specific version.");
+    }
     return queryReader.read(
         resolved,
         owner(granted),
         options,
         body.version(),
-        parseTimestamp("timestamp", body.timestamp()));
+        parseTimestamp("timestamp", body.timestamp()),
+        body.refreshToken(),
+        includeRefreshToken);
   }
 
   private static boolean hasPushdownHint(QueryTableRequest body) {

server/opensharing-server-core/src/main/java/io/opensharing/asset/table/QueryTableRequest.java

@@ -15,8 +15,14 @@ public record QueryTableRequest(
     Long startingVersion,
     Long endingVersion,
     Boolean includeHistoricalProtocol,
-    String idempotencyKey) {
+    String idempotencyKey,
+    Boolean includeRefreshToken,
+    String refreshToken) {
 
   static final QueryTableRequest EMPTY =
-      new QueryTableRequest(null, null, null, null, null, null, null, null, null);
+      new QueryTableRequest(null, null, null, null, null, null, null, null, null, null, null);
+
+  public QueryTableRequest {
+    refreshToken = refreshToken == null || refreshToken.isBlank() ? null : refreshToken;
+  }
 }

server/opensharing-server-core/src/main/java/io/opensharing/asset/table/RefreshTokens.java

@@ -0,0 +1,60 @@
+package io.opensharing.asset.table;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.opensharing.http.ApiException;
+import java.nio.charset.StandardCharsets;
+import java.time.Instant;
+import java.util.Base64;
+
+/** Encodes snapshot refresh tokens so a later query can resign URLs for the same version. */
+public final class RefreshTokens {
+
+  private static final String PREFIX = "os1rt:";
+  private static final ObjectMapper JSON = new ObjectMapper();
+
+  private RefreshTokens() {}
+
+  public record RefreshToken(String id, long version, long expirationTimestamp) {}
+
+  public static String encode(String tableId, long version, Instant expiration) {
+    return encode(new RefreshToken(tableId, version, expiration.toEpochMilli()));
+  }
+
+  public static String encode(RefreshToken token) {
+    try {
+      return Base64.getUrlEncoder()
+          .withoutPadding()
+          .encodeToString(
+              (PREFIX + JSON.writeValueAsString(token)).getBytes(StandardCharsets.UTF_8));
+    } catch (JsonProcessingException e) {
+      throw new IllegalStateException("failed to encode refresh token", e);
+    }
+  }
+
+  public static RefreshToken decode(String token) {
+    try {
+      String decoded =
+          new String(Base64.getUrlDecoder().decode(token.trim()), StandardCharsets.UTF_8);
+      if (!decoded.startsWith(PREFIX)) {
+        throw new IllegalArgumentException("unexpected refresh token");
+      }
+      return JSON.readValue(decoded.substring(PREFIX.length()), RefreshToken.class);
+    } catch (Exception invalid) {
+      throw ApiException.invalidParameter("Error decoding refresh token: " + token + ".");
+    }
+  }
+
+  public static long versionOf(String token, String tableId) {
+    RefreshToken decoded = decode(token);
+    if (decoded.expirationTimestamp() < Instant.now().toEpochMilli()) {
+      throw ApiException.invalidParameter(
+          "The refresh token has expired. Please restart the query.");
+    }
+    if (!tableId.equals(decoded.id())) {
+      throw ApiException.invalidParameter(
+          "The table specified in the refresh token does not match the table being queried.");
+    }
+    return decoded.version();
+  }
+}

server/opensharing-server-core/src/main/java/io/opensharing/asset/table/TableActions.java

@@ -117,6 +117,10 @@ public static String deltaFile(
                 new DeltaSingleAction(add))));
   }
 
+  public static String endStreamAction(String refreshToken, Long minUrlExpirationTimestamp) {
+    return line(new EndStreamLine(new EndStreamAction(refreshToken, minUrlExpirationTimestamp)));
+  }
+
   /**
    * Replace the add path with a signed file URL. When {@code deletionVectorUrl} is set, rewrite the
    * on-disk deletion vector as a path-type descriptor so clients fetch the signed object.
@@ -255,4 +259,9 @@ public record DeltaFile(
 
   public record DeltaSingleAction(
       @JsonSerialize(using = KernelActionSerializer.class) AddFile add) {}
+
+  public record EndStreamLine(EndStreamAction endStreamAction) {}
+
+  @JsonInclude(JsonInclude.Include.NON_NULL)
+  public record EndStreamAction(String refreshToken, Long minUrlExpirationTimestamp) {}
 }

server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/DeltaTableQueryReader.java

@@ -16,6 +16,7 @@
 import io.opensharing.asset.table.DeltaSharingCapabilities;
 import io.opensharing.asset.table.DeltaSharingCapabilities.ResponseFormat;
 import io.opensharing.asset.table.FileIdHash;
+import io.opensharing.asset.table.RefreshTokens;
 import io.opensharing.asset.table.TableActions;
 import io.opensharing.asset.table.signer.SignedUrl;
 import io.opensharing.asset.table.signer.UrlSigners;
@@ -31,7 +32,7 @@
 
 /**
  * Reads Query Table snapshot files from a Delta log. The parquet or delta response is NDJSON:
- * protocol, metaData, then one file action per data file.
+ * protocol, metaData, then one file action per data file, and an endStreamAction when requested.
  *
  * <p>Every file URL is presigned for an hour, or until the catalog's credentials expire if
  * sooner, so the client reads the data files directly from storage without the provider's
@@ -55,18 +56,26 @@ public DeltaTableQueryReader(DeltaKernel kernel, UrlSigners signers) {
    * @param auth the share owner the catalog vends storage credentials for
    * @param version the table version to read; null with a null {@code timestamp} reads the latest
    * @param timestamp read the latest version committed at or before this time
+   * @param refreshToken a refresh token from an earlier response; reads the version it pins
+   * @param includeRefreshToken put a refresh token for this snapshot in the endStreamAction
    */
   public Result read(
       ResolvedAsset table,
       AuthContext auth,
       ResponseOptions options,
       Long version,
-      Instant timestamp) {
+      Instant timestamp,
+      String refreshToken,
+      boolean includeRefreshToken) {
     if (version != null && timestamp != null) {
       throw ApiException.invalidParameter("version and timestamp are mutually exclusive");
     }
     // A time-travel query reports the snapshot version in metaData and on every file.
     boolean historical = version != null || timestamp != null;
+    // A refresh token pins the version it was issued for without making the query historical.
+    if (refreshToken != null) {
+      version = RefreshTokens.versionOf(refreshToken, table.fullName());
+    }
 
     DeltaKernel.Session session = kernel.open(table, auth, URL_TTL);
     Snapshot snapshot = DeltaSnapshots.open(session, version, timestamp);
@@ -94,6 +103,14 @@ public Result read(
       throw new UncheckedIOException(e);
     }
 
+    // A refresh token can only travel in an endStreamAction, so asking for one adds the action.
+    if (includeRefreshToken || options.includeEndStreamAction()) {
+      out.endStreamAction(
+          includeRefreshToken
+              ? RefreshTokens.encode(
+                  table.fullName(), snapshot.getVersion(), Instant.now().plus(URL_TTL))
+              : null);
+    }
     return new Result(snapshot.getVersion(), out.ndjson());
   }
 
@@ -116,11 +133,14 @@ public record Result(long version, String ndjson) {}
    *
    * @param capabilities the request's {@code delta-sharing-capabilities} header; picks the format
    * @param fileIdHash the parsed {@code fileidhash} header, or null to follow the format
+   * @param includeEndStreamAction end the response with an endStreamAction
    */
-  public record ResponseOptions(String capabilities, String fileIdHash) {}
+  public record ResponseOptions(
+      String capabilities, String fileIdHash, boolean includeEndStreamAction) {}
 
   /**
-   * Builds one NDJSON response. File URLs are presigned with the table's storage credentials.
+   * Builds one NDJSON response. File URLs are presigned with the table's storage credentials, and
+   * the earliest expiry among them is reported in the endStreamAction.
    */
   private final class ResponseWriter {
 
@@ -130,6 +150,7 @@ private final class ResponseWriter {
     private final String capabilities;
     private final String fileIdHash;
     private final StringBuilder ndjson = new StringBuilder();
+    private Long minUrlExpirationTimestamp;
 
     ResponseWriter(DeltaKernel.Session session, ResolvedAsset table, ResponseOptions options) {
       this.format = DeltaSharingCapabilities.choose(options.capabilities());
@@ -192,6 +213,11 @@ void snapshotFile(Row scanFile, Long version, Long timestamp) {
       }
     }
 
+    /** {@code refreshToken} may be null. */
+    void endStreamAction(String refreshToken) {
+      ndjson.append(TableActions.endStreamAction(refreshToken, minUrlExpirationTimestamp));
+    }
+
     String ndjson() {
       return ndjson.toString();
     }
@@ -219,6 +245,10 @@ private SignedFile sign(
         deletionVectorFileId = FileIdHash.hash(dvPath, fileIdHash, format);
         expirationTimestamp = Math.min(expirationTimestamp, signedDv.expiration().toEpochMilli());
       }
+      minUrlExpirationTimestamp =
+          minUrlExpirationTimestamp == null
+              ? expirationTimestamp
+              : Math.min(minUrlExpirationTimestamp, expirationTimestamp);
       return new SignedFile(
           file.url(),
           FileIdHash.hash(logPath, fileIdHash, format),

server/opensharing-server-core/src/main/java/io/opensharing/protocol/OpenSharingProtocolService.java

@@ -194,14 +194,19 @@ public HttpResponse queryTable(
     QueryTableRequest query =
         body.isBlank() ? null : ApiJson.MAPPER.readValue(body, QueryTableRequest.class);
     DeltaTableQueryReader.ResponseOptions options =
-        new DeltaTableQueryReader.ResponseOptions(capabilities, FileIdHash.parse(fileIdHash));
+        new DeltaTableQueryReader.ResponseOptions(
+            capabilities,
+            FileIdHash.parse(fileIdHash),
+            DeltaSharingCapabilities.includeEndStreamAction(capabilities));
     DeltaTableQueryReader.Result result =
         tables.query(RecipientTokenDecorator.recipient(ctx), share, schema, table, query, options);
     ResponseHeadersBuilder headers =
         ResponseHeaders.builder(HttpStatus.OK)
             .contentType(NDJSON)
             .add(DELTA_TABLE_VERSION, Long.toString(result.version()))
-            .add(DeltaSharingCapabilities.HEADER, DeltaSharingCapabilities.responded(capabilities));
+            .add(
+                DeltaSharingCapabilities.HEADER,
+                DeltaSharingCapabilities.responded(capabilities, options.includeEndStreamAction()));
     if (options.fileIdHash() != null) {
       headers.add(FileIdHash.HEADER, options.fileIdHash());
     }

server/opensharing-server-core/src/test/java/io/opensharing/asset/table/DeltaSharingCapabilitiesTest.java

@@ -2,7 +2,9 @@
 
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import io.delta.kernel.internal.actions.Protocol;
 import io.opensharing.asset.table.DeltaSharingCapabilities.ResponseFormat;
@@ -19,6 +21,12 @@ void defaultsToParquetWithoutAHeader() {
     assertEquals("responseformat=parquet", DeltaSharingCapabilities.responded(null));
     assertEquals(
         "responseformat=delta", DeltaSharingCapabilities.responded("responseformat=delta"));
+    assertEquals(
+        "responseformat=parquet;includeendstreamaction=true",
+        DeltaSharingCapabilities.responded(null, true));
+    assertFalse(DeltaSharingCapabilities.includeEndStreamAction(null));
+    assertTrue(
+        DeltaSharingCapabilities.includeEndStreamAction("includeendstreamaction=true"));
   }
 
   @Test

server/opensharing-server-core/src/test/java/io/opensharing/asset/table/RefreshTokensTest.java

@@ -0,0 +1,34 @@
+package io.opensharing.asset.table;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import io.opensharing.asset.table.RefreshTokens.RefreshToken;
+import io.opensharing.http.ApiException;
+import java.time.Instant;
+import org.junit.jupiter.api.Test;
+
+class RefreshTokensTest {
+
+  @Test
+  void encodesAndDecodesRefreshToken() {
+    RefreshToken token = new RefreshToken("main.sales.orders", 7, 1000);
+    assertEquals(token, RefreshTokens.decode(RefreshTokens.encode(token)));
+  }
+
+  @Test
+  void roundTripsTableIdAndVersion() {
+    String token = RefreshTokens.encode("main.sales.orders", 7, Instant.now().plusSeconds(60));
+    assertEquals(7, RefreshTokens.versionOf(token, "main.sales.orders"));
+  }
+
+  @Test
+  void rejectsExpiredAndMismatchedTokens() {
+    String expired = RefreshTokens.encode("main.sales.orders", 1, Instant.now().minusSeconds(1));
+    assertThrows(ApiException.class, () -> RefreshTokens.versionOf(expired, "main.sales.orders"));
+    String token = RefreshTokens.encode("main.sales.orders", 1, Instant.now().plusSeconds(60));
+    assertThrows(ApiException.class, () -> RefreshTokens.versionOf(token, "main.hr.salaries"));
+    assertThrows(
+        ApiException.class, () -> RefreshTokens.versionOf("not-a-token", "main.sales.orders"));
+  }
+}

server/opensharing-server-core/src/test/java/io/opensharing/asset/table/TableActionsTest.java

@@ -119,6 +119,21 @@ void encodesParquetAndDeltaFiles() {
         TableActions.deltaFile("file-1", "dv-1", 1L, null, null, signed));
   }
 
+  @Test
+  void encodesEndStreamAction() {
+    assertEquals(
+        """
+        {"endStreamAction":{"refreshToken":"refresh-token",\
+        "minUrlExpirationTimestamp":1652140800000}}
+        """,
+        TableActions.endStreamAction("refresh-token", 1652140800000L));
+    assertEquals(
+        """
+        {"endStreamAction":{}}
+        """,
+        TableActions.endStreamAction(null, null));
+  }
+
   @Test
   void encodesDeltaProtocolAndMetadataFromKernel() {
     Protocol protocol = protocol();