Bump version to 0.8.0-prerelease-1, merge DB migrations for 0.8.0.

Cargo.toml

@@ -20,7 +20,7 @@ homepage = "https://divviup.org"
 license = "MPL-2.0"
 repository = "https://github.com/divviup/janus"
 rust-version = "1.77.0"
-version = "0.7.41"
+version = "0.8.0-prerelease-1"
 
 [workspace.dependencies]
 anyhow = "1"
@@ -50,14 +50,14 @@ hpke-dispatch = "0.7.0"
 http = "1.1"
 http-api-problem = "0.58.0"
 itertools = "0.13"
-janus_aggregator = { version = "0.7.41", path = "aggregator" }
-janus_aggregator_api = { version = "0.7.41", path = "aggregator_api" }
-janus_aggregator_core = { version = "0.7.41", path = "aggregator_core" }
-janus_client = { version = "0.7.41", path = "client" }
-janus_collector = { version = "0.7.41", path = "collector" }
-janus_core = { version = "0.7.41", path = "core" }
-janus_interop_binaries = { version = "0.7.41", path = "interop_binaries" }
-janus_messages = { version = "0.7.41", path = "messages" }
+janus_aggregator = { version = "0.8.0-prerelease-1", path = "aggregator" }
+janus_aggregator_api = { version = "0.8.0-prerelease-1", path = "aggregator_api" }
+janus_aggregator_core = { version = "0.8.0-prerelease-1", path = "aggregator_core" }
+janus_client = { version = "0.8.0-prerelease-1", path = "client" }
+janus_collector = { version = "0.8.0-prerelease-1", path = "collector" }
+janus_core = { version = "0.8.0-prerelease-1", path = "core" }
+janus_interop_binaries = { version = "0.8.0-prerelease-1", path = "interop_binaries" }
+janus_messages = { version = "0.8.0-prerelease-1", path = "messages" }
 k8s-openapi = { version = "0.22.0", features = ["v1_26"] }  # keep this version in sync with what is referenced by the indirect dependency via `kube`
 kube = { version = "0.94.2", default-features = false, features = ["client", "rustls-tls"] }
 mockito = "1.5.0"

aggregator_core/src/datastore.rs

@@ -103,7 +103,7 @@ macro_rules! supported_schema_versions {
 // version is seen, [`Datastore::new`] fails.
 //
 // Note that the latest supported version must be first in the list.
-supported_schema_versions!(7);
+supported_schema_versions!(1);
 
 /// Datastore represents a datastore for Janus, with support for transactional reads and writes.
 /// In practice, Datastore instances are currently backed by a PostgreSQL database.

db/00000000000001_initial_schema.down.sql

@@ -1,12 +1,14 @@
-DROP INDEX outstanding_batches_task_and_time_bucket_index;
+DROP INDEX outstanding_batches_task_id_and_time_bucket_start;
 DROP TABLE outstanding_batches CASCADE;
+DROP TYPE OUTSTANDING_BATCH_STATE CASCADE;
 DROP INDEX aggregate_share_jobs_interval_containment_index CASCADE;
 DROP TABLE aggregate_share_jobs CASCADE;
 DROP INDEX collection_jobs_interval_containment_index CASCADE;
 DROP INDEX collection_jobs_state_and_lease_expiry CASCADE;
 DROP INDEX collection_jobs_task_id_batch_id CASCADE;
 DROP TABLE collection_jobs CASCADE;
 DROP TYPE COLLECTION_JOB_STATE CASCADE;
+DROP INDEX batch_aggregations_gc_time CASCADE;
 DROP TABLE batch_aggregations CASCADE;
 DROP TYPE BATCH_AGGREGATION_STATE;
 DROP INDEX report_aggregations_client_report_id_index CASCADE;

db/00000000000001_initial_schema.up.sql

@@ -31,10 +31,11 @@ CREATE TABLE global_hpke_keys(
 
     -- These columns are mutable.
     state HPKE_KEY_STATE NOT NULL DEFAULT 'PENDING',  -- state of the key
-    updated_at TIMESTAMP NOT NULL,                    -- when the key state was last changed
+    last_state_change_at TIMESTAMP NOT NULL,          -- when the key state was last changed. Used for key rotation logic.
 
     -- creation/update records
     created_at TIMESTAMP NOT NULL,  -- when the row was created
+    updated_at TIMESTAMP NOT NULL,  -- when the row was last changed
     updated_by TEXT NOT NULL        -- the name of the transaction that last updated the row
 );
 
@@ -165,6 +166,21 @@ CREATE TABLE task_upload_counters(
     CONSTRAINT task_upload_counters_unique UNIQUE(task_id, ord)
 ) WITH (fillfactor = 50);
 
+-- Per-task report aggregation counters, used for metrics.
+--
+-- Fillfactor is lowered to improve the likelihood of heap-only tuple optimizations. See the
+-- discussion around this setting for the task_upload_counters table.
+CREATE TABLE task_aggregation_counters(
+    id       BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,  -- artificial ID, internal only
+    task_id  BIGINT NOT NULL,                                  -- task ID the counter is associated with
+    ord      BIGINT NOT NULL,                                  -- the ordinal index of the task aggregation counter
+
+    success  BIGINT NOT NULL DEFAULT 0,  -- reports successfully aggregated
+
+    CONSTRAINT task_aggregation_counters_unique_id UNIQUE(task_id, ord),
+    CONSTRAINT fk_task_id FOREIGN KEY(task_id) REFERENCES tasks(id) ON DELETE CASCADE
+) WITH (fillfactor = 50);
+
 -- The HPKE public keys (aka configs) and private keys used by a given task.
 CREATE TABLE task_hpke_keys(
     id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,  -- artificial ID, internal-only
@@ -321,6 +337,7 @@ CREATE TABLE batch_aggregations(
     CONSTRAINT batch_aggregations_unique_task_id_batch_id_aggregation_param UNIQUE(task_id, batch_identifier, aggregation_param, ord),
     CONSTRAINT fk_task_id FOREIGN KEY(task_id) REFERENCES tasks(id) ON DELETE CASCADE
 );
+CREATE INDEX batch_aggregations_gc_time ON batch_aggregations(task_id, UPPER(COALESCE(batch_interval, client_timestamp_interval)));
 
 -- Specifies the possible state of a collection job.
 CREATE TYPE COLLECTION_JOB_STATE AS ENUM(
@@ -345,6 +362,8 @@ CREATE TABLE collection_jobs(
     helper_aggregate_share     BYTEA,                          -- the helper's encrypted aggregate share (HpkeCiphertext, only if in state FINISHED)
     leader_aggregate_share     BYTEA,                          -- the leader's unencrypted aggregate share (opaque VDAF message, only if in state FINISHED)
 
+    step_attempts  BIGINT NOT NULL DEFAULT 0,  -- the number of attempts to step the collection job without making progress, regardless of whether the lease was successfully released or not
+
     lease_expiry    TIMESTAMP NOT NULL DEFAULT TIMESTAMP '-infinity',  -- when lease on this collection job expires; -infinity implies no current lease
     lease_token     BYTEA,                                             -- a value identifying the current leaseholder; NULL implies no current lease
     lease_attempts  BIGINT NOT NULL DEFAULT 0,                         -- the number of lease acquiries since the last successful lease release
@@ -382,6 +401,12 @@ CREATE TABLE aggregate_share_jobs(
 );
 CREATE INDEX aggregate_share_jobs_interval_containment_index ON aggregate_share_jobs USING gist (task_id, batch_interval);
 
+-- Specifies the possible state of an outstanding batch.
+CREATE TYPE OUTSTANDING_BATCH_STATE AS ENUM(
+    'FILLING',  -- this outstanding batch is still being considered for additional reports
+    'FILLED'    -- this outstanding batch has received enough reports, no more are necessary
+);
+
 -- The leader's view of outstanding batches, which are batches which have not yet started
 -- collection. Used for fixed-size tasks only.
 CREATE TABLE outstanding_batches(
@@ -390,11 +415,13 @@ CREATE TABLE outstanding_batches(
     batch_id BYTEA NOT NULL, -- 32-byte BatchID as defined by the DAP specification.
     time_bucket_start TIMESTAMP,
 
+    state OUTSTANDING_BATCH_STATE NOT NULL DEFAULT 'FILLING',  -- the current state of this outstanding batch
+
     -- creation/update records
     created_at TIMESTAMP NOT NULL,  -- when the row was created
     updated_by TEXT NOT NULL,       -- the name of the transaction that last updated the row
 
     CONSTRAINT outstanding_batches_unique_task_id_batch_id UNIQUE(task_id, batch_id),
     CONSTRAINT fk_task_id FOREIGN KEY(task_id) REFERENCES tasks(id) ON DELETE CASCADE
 );
-CREATE INDEX outstanding_batches_task_and_time_bucket_index ON outstanding_batches (task_id, time_bucket_start);
+CREATE INDEX outstanding_batches_task_id_and_time_bucket_start ON outstanding_batches(task_id, time_bucket_start) WHERE state = 'FILLING';