Register queries and show them from snapshots

PUT /queries/NAME takes a module compiled with aup query: the node loads
and verifies it, checks that its states are the app's (names, types,
records) and that it has a view NAME, and keeps it (64 at most; 409 for
another app's module, 400 for bytes that are no module, 404 without the
view). GET /queries/NAME?from=&count= shows that view from the last
save's snapshot with a reader of the query's module, as a declared view
is shown, so a query runs beside writes and sees one save whole. A query
registered again answers anew.

server_queries registers a per-branch total (a correlated fold), a join
of history rows with their accounts and tellers, each rich account's
rank and a filtered count, checks them against a model of the same
transfers, asks while transfers go on, pages, and checks the refusals.

2 files changed, 270 insertions, 17 deletions

diff --git a/serve/serve.c b/serve/serve.c
index 6467683..588feec 100644
--- a/serve/serve.c
+++ b/serve/serve.c
@@ -69,7 +69,7 @@ typedef struct Host Host;
 /* What a request names: a document, a view (a table's rows from, count) or
  * an input. Kept per work slot, so an accepted event runs even if its
  * connection is gone before the burst. */
-typedef enum { ROUTE_NONE, ROUTE_OBJECT, ROUTE_VIEW, ROUTE_INPUT } RouteKind;
+typedef enum { ROUTE_NONE, ROUTE_OBJECT, ROUTE_VIEW, ROUTE_INPUT, ROUTE_QUERY } RouteKind;
 typedef struct { RouteKind kind; uint32_t target; size_t from, count; } Route;
 
 /* This node's key, for deliveries sealed to it (--key), and whether it
@@ -498,7 +498,7 @@ static bool parse_head(Connection *c, char *head, size_t size, char **query) {
     if (strcmp(version, "HTTP/1.1")) return false;
     if ((*query = strchr(path, '?'))) *(*query)++ = 0;
     static const struct { const char *prefix; RouteKind kind; } routes[] = {
-        {"/objects/", ROUTE_OBJECT}, {"/views/", ROUTE_VIEW}, {"/inputs/", ROUTE_INPUT}};
+        {"/objects/", ROUTE_OBJECT}, {"/views/", ROUTE_VIEW}, {"/inputs/", ROUTE_INPUT}, {"/queries/", ROUTE_QUERY}};
     const char *key = "";
     c->route = ROUTE_NONE;
     for (size_t i = 0; i < sizeof(routes) / sizeof(*routes) && c->route == ROUTE_NONE; ++i)
@@ -514,9 +514,11 @@ static bool parse_head(Connection *c, char *head, size_t size, char **query) {
             return false;
     }
     memcpy(c->key, key, key_size + 1);
-    /* Each route has its method: documents GET and PUT, views GET, inputs POST. */
+    /* Each route has its method: documents GET and PUT, views GET, inputs
+     * POST, queries PUT (registered) and GET (shown). */
     if ((c->route == ROUTE_OBJECT && (c->post || *query)) || (c->route == ROUTE_VIEW && (c->put || c->post)) ||
-        (c->route == ROUTE_INPUT && !c->post)) return false;
+        (c->route == ROUTE_INPUT && !c->post) || (c->route == ROUTE_QUERY && (c->post || (c->put && *query))))
+        return false;
     bool length_seen = false, condition_seen = false, host_seen = false;
     c->source[0] = 0;
     c->sequence = 0;
@@ -586,7 +588,8 @@ static bool parse_head(Connection *c, char *head, size_t size, char **query) {
     }
     if (!*c->source != !c->sequence || (*c->source && c->route != ROUTE_INPUT) || (counted && !*c->source) ||
         c->sequence > UINT64_MAX - c->batch) return false;
-    if (c->put) return host_seen && length_seen && condition_seen;
+    /* A document's PUT names the revision it was made against; a query's replaces what was registered. */
+    if (c->put) return host_seen && length_seen && condition_seen == (c->route != ROUTE_QUERY);
     if (c->post) return host_seen && !condition_seen; /* no Content-Length: no body */
     return host_seen && !condition_seen && c->body_size == 0;
 }
@@ -604,8 +607,27 @@ static const char *resolve(const Host *host, Connection *c, char *query, Route *
     *status = 404;
     c->field_bytes = 0;
     memset(c->given, 0, sizeof(c->given));
-    if (route->kind == ROUTE_NONE || (route->kind == ROUTE_OBJECT && !host->documents)) return "route";
+    if (route->kind == ROUTE_NONE || (route->kind == ROUTE_OBJECT && !host->documents) ||
+        (route->kind == ROUTE_QUERY && !host->owner)) return "route"; /* a query reads an app's states */
     if (route->kind == ROUTE_OBJECT) return NULL;
+    if (route->kind == ROUTE_QUERY) { /* the view is the query module's, found when shown */
+        route->from = 0;
+        route->count = SIZE_MAX;
+        *status = 400;
+        for (char *pair = query; pair && *pair;) {
+            char *end = strchr(pair, '&'), *value = strchr(pair, '=');
+            if (end) *end = 0;
+            uint64_t n;
+            if (!value || (end && value > end)) return "query";
+            *value++ = 0;
+            bool from = !strcmp(pair, "from");
+            if (!number(value, SIZE_MAX, &n) || (!from && strcmp(pair, "count"))) return "query";
+            if (from) route->from = (size_t)n;
+            else route->count = (size_t)n;
+            pair = end ? end + 1 : NULL;
+        }
+        return NULL;
+    }
     int found = route->kind == ROUTE_VIEW ? pure_module_view(m, c->key) : pure_module_input(m, c->key);
     /* An output is the app's to send and a timer ticks by itself: neither is posted. */
     if (found >= 0 && route->kind == ROUTE_INPUT &&
@@ -803,7 +825,8 @@ static void start_request(Connection *c, size_t head_size) {
     size_t input_size = route.kind == ROUTE_INPUT ? lead + c->batch * (table + c->field_bytes) + c->body_size :
                         write ? c->body_size : 0;
     PureResult admitted = pure_runtime_accept(&host->runtime, write ? PURE_INPUT_WRITE : PURE_INPUT_READ, c->key,
-                                              route.kind == ROUTE_INPUT ? 0 : c->expected, input_size, &c->ticket);
+                                              route.kind == ROUTE_INPUT || route.kind == ROUTE_QUERY ? 0 : c->expected,
+                                              input_size, &c->ticket);
     if (admitted != PURE_OK) {
         c->ticket = (PureTicket){0};
         /* An unread body would desynchronize the stream. */
@@ -1461,6 +1484,103 @@ static PureSnapshot *read_published(Host *host) {
     return snapshot;
 }
 
+/* ---- Queries: PUT /queries/NAME registers a module of the app's states and
+ * views of its own (aup query), checked when it comes: it must load (be
+ * verified) with its states laid out as the app's, and have a view NAME.
+ * GET /queries/NAME shows that view from the last save's snapshot, on the
+ * connection thread, as the app's views are: a query never waits for, nor
+ * holds up, the owner's writes. */
+enum { QUERIES = 64 };
+typedef struct { PureModule module; unsigned refs; } Held;
+static struct { char name[65]; Held *held; } queries[QUERIES];
+static pthread_mutex_t queries_lock = PTHREAD_MUTEX_INITIALIZER;
+
+static void let_go_query(Held *held) {
+    pthread_mutex_lock(&queries_lock);
+    bool last = !--held->refs;
+    pthread_mutex_unlock(&queries_lock);
+    if (!last) return;
+    pure_module_release(&held->module);
+    free(held);
+}
+
+/* Whether a's states are laid out as b's: names, kinds, what each refers to,
+ * keys, and the records and document collections behind them. */
+static bool same_states(const PureModule *a, const PureModule *b) {
+    if (a->states != b->states || a->records != b->records || a->record_fields != b->record_fields ||
+        a->collections != b->collections)
+        return false;
+    for (uint32_t i = 0; i < a->states; ++i)
+        if (strcmp(a->state_names[i], b->state_names[i]) || a->state_type[i] != b->state_type[i] ||
+            a->state_ref[i] != b->state_ref[i] || a->state_flags[i] != b->state_flags[i] || a->state_key[i] != b->state_key[i])
+            return false;
+    for (uint32_t i = 0; i < a->records; ++i)
+        if (strcmp(a->record_names[i], b->record_names[i]) || a->record[i].first != b->record[i].first ||
+            a->record[i].count != b->record[i].count)
+            return false;
+    for (uint32_t i = 0; i < a->record_fields; ++i)
+        if (strcmp(a->record_field_names[i], b->record_field_names[i]) || a->record_field_text[i] != b->record_field_text[i])
+            return false;
+    for (uint32_t i = 0; i < a->collections; ++i)
+        if (strcmp(a->collection_names[i], b->collection_names[i])) return false;
+    return true;
+}
+
+static PureResult register_query(const Host *host, const Job *job) {
+    Held *held = calloc(1, sizeof(*held));
+    char error[128];
+    if (!held) return PURE_ALLOCATION;
+    if (pure_module_load(job->effect.value.data, job->effect.value.size, &held->module, error, sizeof(error)) != PURE_OK) {
+        free(held);
+        return PURE_INVALID;
+    }
+    PureResult status = !same_states(&held->module, host->module)          ? PURE_CONFLICT
+                        : pure_module_view(&held->module, job->effect.key) < 0 ? PURE_NOT_FOUND
+                                                                               : PURE_OK;
+    Held *old = NULL;
+    held->refs = 1;
+    if (status == PURE_OK) {
+        pthread_mutex_lock(&queries_lock);
+        int slot = -1;
+        for (int i = 0; i < QUERIES && slot < 0; ++i)
+            if (queries[i].held && !strcmp(queries[i].name, job->effect.key)) slot = i;
+        for (int i = 0; i < QUERIES && slot < 0; ++i)
+            if (!queries[i].held) slot = i;
+        if (slot < 0) status = PURE_LIMIT;
+        else {
+            old = queries[slot].held;
+            snprintf(queries[slot].name, sizeof(queries[slot].name), "%s", job->effect.key);
+            queries[slot].held = held;
+        }
+        pthread_mutex_unlock(&queries_lock);
+    }
+    if (status != PURE_OK) let_go_query(held);
+    if (old) let_go_query(old);
+    return status;
+}
+
+static PureResult run_query(const Job *job, PureSnapshot *snapshot, Applied *a) {
+    Held *held = NULL;
+    pthread_mutex_lock(&queries_lock);
+    for (int i = 0; i < QUERIES && !held; ++i)
+        if (queries[i].held && !strcmp(queries[i].name, job->effect.key)) (held = queries[i].held)->refs++;
+    pthread_mutex_unlock(&queries_lock);
+    if (!held) return PURE_NOT_FOUND;
+    if (!snapshot) {
+        let_go_query(held);
+        return PURE_BUSY;
+    }
+    PureApp reader = {0};
+    pure_app_read(&reader, snapshot);
+    reader.module = &held->module; /* the app's states, the query's views */
+    Route route = job->route;
+    route.target = (uint32_t)pure_module_view(&held->module, job->effect.key);
+    PureResult status = show_view(&reader, &route, job->effect.available, a);
+    pure_app_stop(&reader);
+    let_go_query(held);
+    return status;
+}
+
 /* Every request made ready since the last round: run here, or handed to the
  * owner. Views are shown here from the snapshot when there is one, after the
  * rest went to the owner, which meanwhile applies and saves them. Answering
@@ -1477,31 +1597,39 @@ static void serve_bursts(Host *host) {
             continue;
         }
         PureSnapshot *snapshot = NULL;
-        size_t views = 0;
-        for (size_t i = 0; i < count; ++i) views += host->jobs[i].route.kind == ROUTE_VIEW;
-        if (views && !(snapshot = read_published(host))) views = 0; /* the owner shows them */
-        size_t handed = count - views;
+        size_t views = 0, queried = 0;
+        for (size_t i = 0; i < count; ++i) {
+            views += host->jobs[i].route.kind == ROUTE_VIEW;
+            queried += host->jobs[i].route.kind == ROUTE_QUERY;
+        }
+        if ((views || queried) && !(snapshot = read_published(host))) views = 0; /* the owner shows them */
+        size_t handed = count - views - queried; /* queries are all this thread's */
         if (handed) {
             Owner *owner = host->owner;
             pthread_mutex_lock(&owner->lock);
             if (!grow(&owner->jobs, &owner->capacity, owner->count + handed, sizeof(*owner->jobs))) abort();
             for (size_t i = 0; i < count; ++i)
-                if (!snapshot || host->jobs[i].route.kind != ROUTE_VIEW) owner->jobs[owner->count++] = host->jobs[i];
+                if ((!snapshot || host->jobs[i].route.kind != ROUTE_VIEW) && host->jobs[i].route.kind != ROUTE_QUERY)
+                    owner->jobs[owner->count++] = host->jobs[i];
             owner->arrivals += handed;
             pthread_cond_signal(&owner->waiting);
             pthread_mutex_unlock(&owner->lock);
             host->pending += handed;
         }
-        if (!views) return;
+        if (!views && !queried) return;
         /* Shown first and answered after, as a burst is. */
-        if (!grow(&host->views, &host->views_capacity, views, sizeof(*host->views))) abort();
+        if (!grow(&host->views, &host->views_capacity, views + queried, sizeof(*host->views))) abort();
         size_t shown = 0;
         for (size_t i = 0; i < count; ++i) {
             const Job *job = &host->jobs[i];
-            if (job->route.kind != ROUTE_VIEW) continue;
+            if (job->route.kind != ROUTE_QUERY && (job->route.kind != ROUTE_VIEW || !snapshot)) continue;
             Applied *a = &host->views[shown++];
-            *a = (Applied){.host = host, .ticket = job->effect.ticket, .input = job->effect.input, .route = ROUTE_VIEW};
-            a->status = show_view(&host->reader, &job->route, job->effect.available, a);
+            bool registering = job->route.kind == ROUTE_QUERY && job->effect.input == PURE_INPUT_WRITE;
+            *a = (Applied){.host = host, .ticket = job->effect.ticket, .input = job->effect.input,
+                           .route = registering ? ROUTE_QUERY : ROUTE_VIEW};
+            a->status = job->route.kind == ROUTE_VIEW ? show_view(&host->reader, &job->route, job->effect.available, a)
+                        : registering              ? register_query(host, job)
+                                                   : run_query(job, snapshot, a);
         }
         pure_snapshot_release(snapshot); /* what answers lend holds its own mapping */
         for (size_t i = 0; i < shown; ++i) answer_applied(host, &host->views[i]);
diff --git a/tests/check.py b/tests/check.py
index 9ff771a..d262b8e 100644
--- a/tests/check.py
+++ b/tests/check.py
@@ -1023,6 +1023,130 @@ def server_surface(aup, pure, work):
     return {'events': 5, 'lines': len(got)}
 
 
+def server_queries(aup, pure, work):
+    """Queries as verified programs over a snapshot (PUT and GET /queries/NAME):
+    aup query compiles a bank's declarations with views of its own; the node
+    checks the module when it is registered (it loads, its states are the
+    bank's, it has the view named) and shows that view from the last save's
+    snapshot. A per-branch total (a fold reading its column's row, as a
+    correlated subquery does), a join of history rows with their accounts and
+    tellers (keyed lookups), and a self-join (each rich account's rank) match
+    a model of the same events, also asked while transfers go on; a module of
+    another app is refused (409), bytes that are no module too (400), and a
+    query registered again answers anew."""
+    import random
+    import threading
+    bank = pure / 'benchmarks' / 'bank' / 'bank.aup'
+    module = work / 'queries-bank.pmod'
+    compile_module(aup, bank, module)
+    queries = {
+        'per_branch': 'view per_branch = each b in branches {\n  bid = b.bid\n'
+                      '  total = sum(a.balance for a in accounts where a.bid == b.bid)\n'
+                      '  accounts = count(a in accounts where a.bid == b.bid)\n}\n',
+        'large': 'view large = each h in history where h.delta > 4000 {\n  aid = h.aid\n  delta = h.delta\n'
+                 '  account = accounts[h.aid].balance\n  teller = tellers[h.tid].balance\n}\n',
+        'ranks': 'view ranks = each a in accounts where a.balance > 15000 {\n  aid = a.aid\n'
+                 '  richer = count(b in accounts where b.balance > a.balance)\n}\n',
+        'rich': 'view rich = count(a in accounts where a.balance > 0)\n',
+    }
+    compiled = {}
+    for name, text in queries.items():
+        source = work / f'query-{name}.aup'
+        source.write_text(text)
+        out = work / f'query-{name}.pmod'
+        run(aup, 'query', bank, source, '-o', out)
+        compiled[name] = out.read_bytes()
+    other = work / 'queries-other.pmod'
+    (work / 'queries-other.aup').write_text('state accounts = 0\nview rich = accounts\n')
+    compile_module(aup, work / 'queries-other.aup', other)
+    host = HttpHost([pure / 'build/pure-serve', '--module', module, '--store', work / 'queries-store', '--threads', '2'])
+
+    def call(method, path, body=None, json_body=False):
+        import http.client
+        connection = http.client.HTTPConnection('127.0.0.1', host.port, timeout=30)
+        try:
+            connection.request(method, path, body, {'Content-Type': 'application/json'} if json_body else {})
+            answer = connection.getresponse()
+            return answer.status, answer.getheader('X-Pure-Error'), answer.read()
+        finally:
+            connection.close()
+
+    try:
+        for b in range(1, 4):
+            assert call('POST', '/inputs/branch', json.dumps({'bid': b}).encode(), True)[0] == 200
+        for t in range(1, 31):
+            assert call('POST', '/inputs/teller', json.dumps({'tid': t, 'bid': (t - 1) // 10 + 1}).encode(), True)[0] == 200
+        for a in range(1, 301):
+            assert call('POST', '/inputs/account', json.dumps({'aid': a, 'bid': (a - 1) // 100 + 1}).encode(), True)[0] == 200
+        accounts, tellers, history = {a: 0 for a in range(1, 301)}, {t: 0 for t in range(1, 31)}, []
+        rng = random.Random(8)
+
+        def transfer():
+            aid, tid, bid, delta = rng.randint(1, 300), rng.randint(1, 30), rng.randint(1, 3), rng.randint(-5000, 5000)
+            assert call('POST', '/inputs/transfer', json.dumps({'aid': aid, 'tid': tid, 'bid': bid, 'delta': delta}).encode(),
+                        True)[0] == 200
+            accounts[aid] += delta
+            tellers[tid] += delta
+            history.append((aid, tid, delta))
+
+        for _ in range(600):
+            transfer()
+        for name, bytes_ in compiled.items():
+            assert call('PUT', f'/queries/{name}', bytes_) == (200, 'none', b''), name
+
+        def expected(name):
+            if name == 'per_branch':
+                return [{'bid': b, 'total': sum(v for a, v in accounts.items() if (a - 1) // 100 + 1 == b), 'accounts': 100}
+                        for b in range(1, 4)]
+            if name == 'large':
+                return [{'aid': a, 'delta': d, 'account': accounts[a], 'teller': tellers[t]} for a, t, d in history if d > 4000]
+            if name == 'ranks':
+                return [{'aid': a, 'richer': sum(1 for v in accounts.values() if v > balance)}
+                        for a, balance in sorted(accounts.items()) if balance > 15000]
+            return sum(1 for v in accounts.values() if v > 0)
+
+        def answer(name):
+            status, error, data = call('GET', f'/queries/{name}')
+            assert status == 200, (name, status, error)
+            return json.loads(data)['items'] if name != 'rich' else int(data)
+
+        for name in compiled:
+            assert answer(name) == expected(name), name
+        # Asked while transfers go on, a query sees one save or the next, whole.
+        seen, stop = [], threading.Event()
+
+        def ask():
+            while not stop.is_set():
+                seen.append(sum(row['total'] for row in answer('per_branch')))
+
+        asker = threading.Thread(target=ask)
+        asker.start()
+        for _ in range(200):
+            transfer()
+        stop.set()
+        asker.join()
+        totals = [0]
+        for _, _, delta in history:
+            totals.append(totals[-1] + delta)
+        assert seen and all(total in totals for total in seen), seen[:5]
+        assert answer('per_branch') == expected('per_branch')
+        page = json.loads(call('GET', '/queries/large?from=1&count=2')[2])
+        assert page['items'] == expected('large')[1:3] and page['rows'] == len(expected('large'))
+        # Refused: another app's states, bytes that are no module, a module without the view named.
+        assert call('PUT', '/queries/rich', other.read_bytes())[:2] == (409, 'conflict')
+        assert call('PUT', '/queries/rich', b'not a module')[:2] == (400, 'invalid')
+        assert call('PUT', '/queries/missing', compiled['rich'])[:2] == (404, 'not_found')
+        assert call('GET', '/queries/missing')[:2] == (404, 'no_value')
+        # Registered again: the new module answers.
+        (work / 'query-rich2.aup').write_text('view rich = count(a in accounts where a.balance < 0)\n')
+        run(aup, 'query', bank, work / 'query-rich2.aup', '-o', work / 'query-rich2.pmod')
+        assert call('PUT', '/queries/rich', (work / 'query-rich2.pmod').read_bytes())[0] == 200
+        assert answer('rich') == sum(1 for v in accounts.values() if v < 0)
+    finally:
+        host.stop()
+    return {'queries': len(compiled), 'asked_while_writing': len(seen), 'history': len(history)}
+
+
 def server_webhook(aup, pure, work):
     """An output wired to a third-party HTTPS endpoint (a Python server with
     a certificate of its own): five orders arrive in order as JSON with the
@@ -1483,6 +1607,7 @@ def main():
         report['server_webhook'] = server_webhook(aup, pure, work)
         report['server_sealed'] = server_sealed(aup, pure, work)
         report['server_surface'] = server_surface(aup, pure, work)
+        report['server_queries'] = server_queries(aup, pure, work)
         report['server_answers'] = server_answers(aup, pure, work)
     print(json.dumps(report))