Skip to content

Commit f09a512

Browse files
rustyconoverclaude
andcommitted
Fix named-arg constraints, unanchored pattern, HTTP token key, listing binds
- ConstraintEnforcer: named-only specs (position -1) are looked up by name and no longer consume a positional slot, so a named arg is checked against its own value and the positional consts after it read the right slot. Varargs consts check every value. - pattern is an unanchored search (Matcher.find), matching JSON Schema/ECMA and the Rust and Go SDKs; anchor with ^/$ for a full match. - HTTP: when no tokenKey is configured, the stream-state token key is derived from the signing key (HKDF-SHA256, label vgi.http.state_token_key.v1), so table-function streams resume across replicas and restarts. Without a signing key the per-process key stays and a warning is logged. - Catalog listing: an onBind that throws on the empty-argument probe no longer breaks ATTACH; the function is advertised with an empty output schema and validated at the real bind. Bump to 0.46.1. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1 parent cfa5783 commit f09a512

14 files changed

Lines changed: 521 additions & 28 deletions

File tree

‎README.md‎

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -45,7 +45,7 @@ Artifacts are published to Maven Central under the `farm.query` group.
4545

4646
```kotlin
4747
dependencies {
48-
implementation("farm.query:vgi:0.46.0")
48+
implementation("farm.query:vgi:0.46.1")
4949
}
5050
```
5151

@@ -55,7 +55,7 @@ dependencies {
5555
<dependency>
5656
<groupId>farm.query</groupId>
5757
<artifactId>vgi</artifactId>
58-
<version>0.46.0</version>
58+
<version>0.46.1</version>
5959
</dependency>
6060
```
6161

@@ -143,6 +143,16 @@ SELECT demo.multiply(21, 2); -- 42
143143

144144
The `launch:` location scheme starts the worker once behind a flock-coordinated Unix socket and reuses it across queries and engine processes — essential for JVM workers, which are expensive to cold-start. Plain subprocess (`/path/to/worker`) and `http(s)://` locations also work.
145145

146+
## Running several HTTP replicas
147+
148+
Set the same `VGI_SIGNING_KEY` (or call `Worker.signingKey(byte[])`) on every
149+
replica. It seals `attach_opaque_data` and, through an HKDF-SHA256 derivation
150+
(label `vgi.http.state_token_key.v1`, never the raw key), keys the HTTP
151+
stream-state tokens, so a table-function stream started on one replica resumes
152+
on any other and survives a restart. Without it each process uses a random key
153+
and logs a warning at startup. An explicit `HttpServer.Config.Builder.tokenKey`
154+
still takes precedence.
155+
146156
## Example worker
147157

148158
For authenticated `iroh://` and HTTP-semantics `httpi://` deployments, see

‎build.gradle.kts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@ plugins {
88

99
allprojects {
1010
group = "farm.query"
11-
version = "0.46.0"
11+
version = "0.46.1"
1212

1313
repositories {
1414
mavenCentral()

‎examples/docs/README.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -49,5 +49,5 @@ These build against `project(":vgi")` so they track the SDK at HEAD. A reader
4949
does the same thing with one line:
5050

5151
```kotlin
52-
implementation("farm.query:vgi:0.46.0")
52+
implementation("farm.query:vgi:0.46.1")
5353
```

‎examples/docs/build.gradle.kts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@
1010
// the SDK at HEAD and a breaking change shows up here rather than in a reader's
1111
// editor. A reader does the same thing with one line instead:
1212
//
13-
// implementation("farm.query:vgi:0.46.0")
13+
// implementation("farm.query:vgi:0.46.1")
1414
//
1515
// which is what the documentation shows. The toolchain, `-parameters` and the
1616
// JVM opens all come from the root build, so nothing else is repeated here.

‎vgi/src/main/java/farm/query/vgi/Worker.java‎

Lines changed: 90 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -496,7 +496,9 @@ public Worker maxAuthAge(double seconds) {
496496
*
497497
* <p>Any length: a 32-byte key is used as-is, any other is replaced by its SHA-256, exactly as
498498
* {@code VGI_SIGNING_KEY} is normalized in every VGI SDK. A configured key also lets replicas
499-
* open each other's opaque data and tickets. Without one each HTTP process seals with a random
499+
* open each other's opaque data and tickets, and keys the HTTP stream-state tokens (an
500+
* HKDF-SHA256 derivation, never the raw key; see {@link #runHttp}) so table-function streams
501+
* resume on any replica and across restarts. Without one each HTTP process seals with a random
500502
* key of its own, and attach tickets are not hosted: every ticket would die on restart.
501503
*
502504
* @param key the key bytes, or {@code null} to defer to {@code VGI_SIGNING_KEY}
@@ -2265,6 +2267,12 @@ public void runFromArgs(String[] args) {
22652267
* (15 s by default; see {@code HttpServer}'s {@code setStopTimeout}) and
22662268
* then forcibly closes any stragglers.
22672269
*
2270+
* <p>State tokens: when {@code config} sets no {@code tokenKey} and a signing key is
2271+
* configured ({@link #signingKey(byte[])} or {@code VGI_SIGNING_KEY}), the stream-state
2272+
* token key is derived from it (HKDF-SHA256, label {@value #STATE_TOKEN_KEY_LABEL}), so
2273+
* every replica sharing the signing key can resume table-function streams another replica
2274+
* started. With neither, the key is random per process and a warning is logged.
2275+
*
22682276
* @param config fully-built HTTP server configuration
22692277
* @throws Exception if the server fails to start or serve
22702278
*/
@@ -2276,6 +2284,7 @@ public void runHttp(HttpServer.Config config) throws Exception {
22762284
HttpServer.Config effective = config.landingInfo() != null
22772285
? config
22782286
: config.withLandingInfo(farm.query.vgi.http.WorkerLandingInfo.of(this));
2287+
effective = withStateTokenKey(effective, configuredSigningKey(), log);
22792288
HttpServer http = new HttpServer(buildServer(Transport.HTTP), effective);
22802289
http.start();
22812290
System.out.println("PORT:" + http.port());
@@ -2294,6 +2303,86 @@ public void runHttp(HttpServer.Config config) throws Exception {
22942303
http.join();
22952304
}
22962305

2306+
/** HKDF {@code info} label for the HTTP stream-state token key derived from the signing key. */
2307+
static final String STATE_TOKEN_KEY_LABEL = "vgi.http.state_token_key.v1";
2308+
2309+
/**
2310+
* Give an HTTP config a stable stream-state token key when it has none.
2311+
*
2312+
* <p>Table-function state (cursors, the per-turn producer state) travels between HTTP
2313+
* requests in AEAD-sealed state tokens. With no token key vgi-rpc seals them with a random
2314+
* per-process key, so a token minted by one replica (or before a restart) cannot be opened by
2315+
* another. When a signing key is configured ({@link #signingKey(byte[])} or
2316+
* {@code VGI_SIGNING_KEY}) the token key is derived from it with HKDF-SHA256 under the label
2317+
* {@value #STATE_TOKEN_KEY_LABEL} (never the raw signing key), so every replica sharing the
2318+
* signing key can resume every other replica's streams. An explicit
2319+
* {@code HttpServer.Config.Builder#tokenKey} always wins. Without either, the per-process key
2320+
* is kept and a one-line warning is logged.
2321+
*
2322+
* @param config the HTTP config
2323+
* @param signingKey the normalized configured signing key, or {@code null}
2324+
* @param log where to log the single-replica warning, or {@code null}
2325+
* @return {@code config}, or a copy carrying the derived token key
2326+
*/
2327+
static HttpServer.Config withStateTokenKey(HttpServer.Config config, byte[] signingKey,
2328+
org.slf4j.Logger log) {
2329+
if (config.tokenKey() != null) return config;
2330+
if (signingKey == null) {
2331+
if (log != null) {
2332+
log.warn("HTTP state tokens are sealed with a random per-process key: table-function "
2333+
+ "streams cannot resume on another replica or after a restart; set "
2334+
+ SIGNING_KEY_ENV + " (or Worker.signingKey) for multi-replica deployments");
2335+
}
2336+
return config;
2337+
}
2338+
return copyConfigWithTokenKey(config, deriveStateTokenKey(signingKey));
2339+
}
2340+
2341+
/**
2342+
* Derive the 32-byte HTTP state-token key from a signing key: HKDF-SHA256 (RFC 5869) with an
2343+
* all-zero salt and {@code info} = {@value #STATE_TOKEN_KEY_LABEL}.
2344+
*
2345+
* @param signingKey the signing key (input keying material)
2346+
* @return the derived 32-byte key
2347+
*/
2348+
static byte[] deriveStateTokenKey(byte[] signingKey) {
2349+
try {
2350+
javax.crypto.Mac mac = javax.crypto.Mac.getInstance("HmacSHA256");
2351+
mac.init(new javax.crypto.spec.SecretKeySpec(new byte[32], "HmacSHA256"));
2352+
byte[] prk = mac.doFinal(signingKey);
2353+
mac.init(new javax.crypto.spec.SecretKeySpec(prk, "HmacSHA256"));
2354+
mac.update(STATE_TOKEN_KEY_LABEL.getBytes(java.nio.charset.StandardCharsets.UTF_8));
2355+
mac.update((byte) 1);
2356+
return mac.doFinal(); // L = 32 = HashLen: one expand block
2357+
} catch (java.security.GeneralSecurityException e) {
2358+
throw new IllegalStateException("HmacSHA256 unavailable", e);
2359+
}
2360+
}
2361+
2362+
/** Copy the {@code HttpServer.Config} record with {@code tokenKey} replaced, via its canonical
2363+
* constructor, so this keeps working as vgi-rpc adds config components. */
2364+
private static HttpServer.Config copyConfigWithTokenKey(HttpServer.Config config, byte[] tokenKey) {
2365+
try {
2366+
java.lang.reflect.RecordComponent[] comps = HttpServer.Config.class.getRecordComponents();
2367+
Object[] values = new Object[comps.length];
2368+
Class<?>[] types = new Class<?>[comps.length];
2369+
boolean found = false;
2370+
for (int i = 0; i < comps.length; i++) {
2371+
types[i] = comps[i].getType();
2372+
if ("tokenKey".equals(comps[i].getName())) {
2373+
values[i] = tokenKey;
2374+
found = true;
2375+
} else {
2376+
values[i] = comps[i].getAccessor().invoke(config);
2377+
}
2378+
}
2379+
if (!found) throw new IllegalStateException("HttpServer.Config has no tokenKey component");
2380+
return HttpServer.Config.class.getDeclaredConstructor(types).newInstance(values);
2381+
} catch (ReflectiveOperationException e) {
2382+
throw new IllegalStateException("cannot set the HTTP state-token key", e);
2383+
}
2384+
}
2385+
22972386
private static void requireLoopback(String host, String label) {
22982387
if (!("127.0.0.1".equals(host) || "::1".equals(host) || "localhost".equals(host))) {
22992388
throw new IllegalArgumentException(label + " must bind loopback, got " + host);

‎vgi/src/main/java/farm/query/vgi/function/ArgSpec.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -78,7 +78,7 @@ public record ArgSpec(
7878
* @param le inclusive upper bound ({@code value <= le}), or {@code null}.
7979
* @param gt exclusive lower bound ({@code value > gt}), or {@code null}.
8080
* @param lt exclusive upper bound ({@code value < lt}), or {@code null}.
81-
* @param pattern regex the value must match, or {@code null}.
81+
* @param pattern regex the value must contain a match for (unanchored search; anchor with {@code ^}/{@code $} for a full match), or {@code null}.
8282
*/
8383
public record Constraints(
8484
List<Object> choices,
@@ -132,7 +132,7 @@ public static Constraints choices(List<Object> choices) {
132132
/**
133133
* Regex constraint only.
134134
*
135-
* @param pattern the regex the value must match.
135+
* @param pattern the regex searched for in the value (unanchored; anchor with {@code ^}/{@code $} for a full match).
136136
* @return a pattern-only {@link Constraints}.
137137
*/
138138
public static Constraints pattern(String pattern) {

‎vgi/src/main/java/farm/query/vgi/function/ConstraintEnforcer.java‎

Lines changed: 41 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -24,24 +24,56 @@ private ConstraintEnforcer() {}
2424
/**
2525
* Validate a function's const arguments against their declared constraints.
2626
*
27-
* <p>Const arguments are numbered sequentially (the i-th const spec reads the
28-
* i-th positional value); a value that violates a declared choices /
29-
* ge/le/gt/lt / pattern constraint throws {@link IllegalArgumentException}. A
30-
* null (absent) value is skipped, matching the other SDKs.
27+
* <p>Where each const argument's value is read from:
28+
* <ul>
29+
* <li>a <b>named-only</b> spec ({@code position() < 0}, e.g.
30+
* {@link ArgSpec#named}) reads {@code args.named().get(name)} and does
31+
* <em>not</em> consume a positional slot;</li>
32+
* <li>a <b>positional</b> const spec reads the next value of
33+
* {@code args.positional()}: the wire carries only the const values
34+
* there, densely, in declaration order (column, TABLE and named
35+
* arguments are not in it), so the i-th positional const spec reads
36+
* the i-th positional value;</li>
37+
* <li>a <b>varargs</b> const spec checks every remaining positional value.</li>
38+
* </ul>
39+
* A value that violates a declared choices / ge/le/gt/lt / pattern constraint
40+
* throws {@link IllegalArgumentException}. A null (absent) value is skipped,
41+
* matching the other SDKs, so a catalog-listing bind with empty arguments
42+
* never trips a constraint.
43+
*
44+
* <p>{@code pattern} is an <b>unanchored search</b> (like JSON Schema / ECMA
45+
* {@code pattern}, and the Rust and Go SDKs): {@code "https?://"} accepts
46+
* {@code "see http://x"}. Anchor explicitly with {@code ^} / {@code $} to
47+
* require a full match.
3148
*
3249
* @param args the bound call arguments (may be {@code null}).
3350
* @param specs the function's argument specs in declaration order (may be {@code null}).
3451
*/
3552
public static void enforce(Arguments args, List<ArgSpec> specs) {
3653
if (args == null || specs == null) return;
54+
List<Object> positional = args.positional() == null ? List.of() : args.positional();
55+
java.util.Map<String, Object> named = args.named() == null ? java.util.Map.of() : args.named();
3756
int constIdx = 0;
3857
for (ArgSpec spec : specs) {
3958
if (!spec.isConst()) continue;
4059
ArgSpec.Constraints c = spec.constraints();
41-
if (c != null && !c.isEmpty()) {
42-
Object value = constIdx < args.positional().size()
43-
? args.positional().get(constIdx)
44-
: null;
60+
boolean constrained = c != null && !c.isEmpty();
61+
if (spec.position() < 0) {
62+
// Named-only: looked up by name, consumes no positional slot.
63+
if (constrained) validateConstValue(spec.name(), c, named.get(spec.name()));
64+
continue;
65+
}
66+
if (spec.varargs()) {
67+
if (constrained) {
68+
for (int i = constIdx; i < positional.size(); i++) {
69+
validateConstValue(spec.name(), c, positional.get(i));
70+
}
71+
}
72+
constIdx = Math.max(constIdx, positional.size());
73+
continue;
74+
}
75+
if (constrained) {
76+
Object value = constIdx < positional.size() ? positional.get(constIdx) : null;
4577
validateConstValue(spec.name(), c, value);
4678
}
4779
constIdx++;
@@ -70,7 +102,7 @@ private static void validateConstValue(String name, ArgSpec.Constraints c, Objec
70102
}
71103
if (c.pattern() != null && !c.pattern().isEmpty() && value instanceof String s) {
72104
try {
73-
if (!Pattern.matches(c.pattern(), s)) {
105+
if (!Pattern.compile(c.pattern()).matcher(s).find()) {
74106
throw constraintError(name, "must match pattern " + c.pattern());
75107
}
76108
} catch (PatternSyntaxException ignored) {

‎vgi/src/main/java/farm/query/vgi/internal/ArgumentSpecSerializer.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,7 @@
3838
* <li>{@code vgi_choices} — JSON array of the closed set of allowed values.</li>
3939
* <li>{@code vgi_range} — interval notation built from {@code ge}/{@code le}/
4040
* {@code gt}/{@code lt} (e.g. {@code "[0, 10]"}, {@code "(0, +inf)"}).</li>
41-
* <li>{@code vgi_pattern} — the raw regex the value must match.</li>
41+
* <li>{@code vgi_pattern} — the raw regex, an unanchored search over the value.</li>
4242
* </ul>
4343
*/
4444
public final class ArgumentSpecSerializer {

‎vgi/src/main/java/farm/query/vgi/internal/VgiServiceImpl.java‎

Lines changed: 35 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -3224,8 +3224,9 @@ private FunctionInfo toScalarFunctionInfo(ScalarFunction fn, String schemaName)
32243224
* @return the wire {@link FunctionInfo}.
32253225
*/
32263226
static FunctionInfo scalarFunctionInfo(ScalarFunction fn, String schemaName) {
3227-
BindResponse r = fn.onBind(new ScalarBindParams(fn.name(), Arguments.empty(), null, Map.of()));
3228-
return baseFunctionInfo(fn, schemaName, "SCALAR", bindOutput(r), false);
3227+
byte[] out = catalogBindOutput(fn.name(),
3228+
() -> fn.onBind(new ScalarBindParams(fn.name(), Arguments.empty(), null, Map.of())));
3229+
return baseFunctionInfo(fn, schemaName, "SCALAR", out, false);
32293230
}
32303231

32313232
private FunctionInfo toTableFunctionInfo(TableFunction fn, String schemaName) {
@@ -3251,20 +3252,22 @@ static FunctionInfo tableFunctionInfo(TableFunction fn, String schemaName) {
32513252
return withRequiredSecrets(baseFunctionInfo(fn, schemaName, "TABLE",
32523253
SchemaUtil.serializeSchema(new Schema(List.of())), false), fn.requiredSecrets());
32533254
}
3254-
BindResponse r = fn.onBind(new TableBindParams(fn.name(), Arguments.empty(), null, Map.of()));
3255-
return withRequiredSecrets(baseFunctionInfo(fn, schemaName, "TABLE", bindOutput(r), false),
3255+
byte[] out = catalogBindOutput(fn.name(),
3256+
() -> fn.onBind(new TableBindParams(fn.name(), Arguments.empty(), null, Map.of())));
3257+
return withRequiredSecrets(baseFunctionInfo(fn, schemaName, "TABLE", out, false),
32563258
fn.requiredSecrets());
32573259
}
32583260

32593261
private FunctionInfo toTableInOutFunctionInfo(TableInOutFunction fn, String schemaName) {
3260-
BindResponse r = fn.onBind(new TableInOutBindParams(fn.name(), Arguments.empty(), null, Map.of()));
3262+
byte[] out = catalogBindOutput(fn.name(),
3263+
() -> fn.onBind(new TableInOutBindParams(fn.name(), Arguments.empty(), null, Map.of())));
32613264
// has_finalize drives the C++ per-substream FINALIZE-phase init after
32623265
// input EOS (parallel streaming finalize). Blended (RowTransformFunction)
32633266
// registrations additionally advertise input_from_args, entering the
32643267
// C++ in-out registration branch with real-typed positional args; the
32653268
// C++ rejects input_from_args && has_finalize, and RowTransformFunction
32663269
// pins hasFinalize() to false, so the pair stays consistent.
3267-
FunctionInfo base = baseFunctionInfo(fn, schemaName, "TABLE", bindOutput(r), fn.hasFinalize());
3270+
FunctionInfo base = baseFunctionInfo(fn, schemaName, "TABLE", out, fn.hasFinalize());
32683271
// Re-stamp the TIO-specific fields baseFunctionInfo hardcodes:
32693272
// max_workers (0 = the C++ parallel-by-default per-substream fan-out;
32703273
// 1 would be the A3 serial opt-out AND disqualify the exchange-mode
@@ -3325,9 +3328,10 @@ private static FunctionInfo withRequiredSecrets(
33253328

33263329
private FunctionInfo toBufferingFunctionInfo(
33273330
farm.query.vgi.buffering.TableBufferingFunction fn, String schemaName) {
3328-
BindResponse r = fn.onBind(new TableInOutBindParams(fn.name(), Arguments.empty(), null, Map.of()));
3331+
byte[] out = catalogBindOutput(fn.name(),
3332+
() -> fn.onBind(new TableInOutBindParams(fn.name(), Arguments.empty(), null, Map.of())));
33293333
FunctionInfo base = baseFunctionInfo(fn.name(), fn.metadata(), schemaName, "TABLE_BUFFERING",
3330-
ArgumentSpecSerializer.toIpcBytes(fn.argumentSpecs()), bindOutput(r),
3334+
ArgumentSpecSerializer.toIpcBytes(fn.argumentSpecs()), out,
33313335
encodeParameterDefaults(fn), /*hasFinalize=*/true, 1, null);
33323336
// Re-stamp the buffering-specific ordering flags (baseFunctionInfo
33333337
// hardcodes them false; they're only meaningful for TableBuffering).
@@ -3353,6 +3357,29 @@ private static byte[] bindOutput(BindResponse r) {
33533357
return r.output_schema() != null ? r.output_schema() : new byte[0];
33543358
}
33553359

3360+
private static final org.slf4j.Logger CATALOG_LOG =
3361+
org.slf4j.LoggerFactory.getLogger(VgiServiceImpl.class);
3362+
3363+
/**
3364+
* Run a catalog-listing bind (empty arguments, no input schema) and return its
3365+
* output schema bytes. A function whose {@code onBind} rejects the empty-argument
3366+
* probe (validating required arguments, needing an attach, ...) must not break
3367+
* ATTACH for the whole catalog: the failure is swallowed and an empty output
3368+
* schema is advertised instead, which is what vgi-python always advertises for
3369+
* table functions. The real bind at call time still runs {@code onBind} with the
3370+
* caller's arguments. Constraint enforcement is a no-op here: every declared
3371+
* argument is absent, and absent values skip their constraints.
3372+
*/
3373+
private static byte[] catalogBindOutput(String name, java.util.function.Supplier<BindResponse> bind) {
3374+
try {
3375+
return bindOutput(bind.get());
3376+
} catch (RuntimeException e) {
3377+
CATALOG_LOG.debug("catalog listing: onBind of '{}' with empty arguments threw ({}); "
3378+
+ "advertising an empty output schema", name, e.toString());
3379+
return SchemaUtil.serializeSchema(new Schema(List.of()));
3380+
}
3381+
}
3382+
33563383
private static farm.query.vgi.pushdown.PushdownFiltersDecoder.Capabilities filterCapabilities(
33573384
FunctionMetadata metadata) {
33583385
Set<farm.query.vgi.pushdown.FilterIdentity> functions = new java.util.LinkedHashSet<>();

0 commit comments

Comments
 (0)