Repository: google/exposure-notifications-server
Pull request #1119 (2020-10-29 01:25:58)

write generated export padding to the database, ensure overlapping ex…
…ports contain same data

## Proposed Changes

* Ensures that exports with overlapping time periods contain the same generated data by persisting the data to the database
* Obtain a lock over export regions when creating exports, all region locks are required, and they are sorted and obtained in order
* Batches are attempted to be processed in an order such that padding for earlier batches should be generated and thus included in later batches

**Release Note**

```release-note
Export padding is persisted to the database to ensure stability over time.
```

google-oss-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/google/exposure-notifications-server/pull/1119#" title="Author self-approved">mikehelmick</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=google%2Fexposure-notifications-server).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/google/exposure-notifications-server/blob/main/OWNERS)~~ [mikehelmick]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
mikehelmick: ready to review but putting a 
/hold
while I check on some things

Patch:
diff --git a/RELEASING.md b/RELEASING.md
index 9e927a5..3800b7f 100644
--- a/RELEASING.md
+++ b/RELEASING.md
@@ -1,9 +1,12 @@
 # Exposure Notifications Servers Release Process
 
-  * Releases are cut every Tuesday (target time 9am, US Pacific Time)
+  * Releases are cut every other Tuesday (target time 9am, US Pacific Time)
   * All releases will have a release branch and a corresponding tag
   * Both the exposure-notifications-sever and exposure-notifications-verification-server will follow the same release numbering
 
+Most issues will not be backported to previous releases. It is recommended that server
+operators keep up with the release cadence of the open source project as much as possible.
+
 # Dependencies
 
 You need the kubernetes release-notes tool to generate release notes.
diff --git a/internal/database/lock.go b/internal/database/lock.go
index 856d189..dde6751 100644
--- a/internal/database/lock.go
+++ b/internal/database/lock.go
@@ -18,6 +18,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"sort"
 	"time"
 
 	"github.com/google/exposure-notifications-server/pkg/logging"
@@ -35,6 +36,59 @@ var (
 // UnlockFn can be deferred to release a lock.
 type UnlockFn func() error
 
+// MultiLock obtains multiple locks in a single transaction. Either all locks are obtained, or
+// the transaction is rolled back.
+// The lockIDs are sorted by normal ascending string sort order before obtaining the locks.
+func (db *DB) MultiLock(ctx context.Context, lockIDs []string, ttl time.Duration) (UnlockFn, error) {
+	if len(lockIDs) == 0 {
+		return nil, fmt.Errorf("no lockIDs presented")
+	}
+
+	lockOrder := make([]string, len(lockIDs))
+	// Make a copy of the slice so that we have a stable slice in the unlcok function.
+	copy(lockOrder, lockIDs)
+	sort.Strings(lockOrder)
+
+	var expires time.Time
+	err := db.InTx(ctx, pgx.Serializable, func(tx pgx.Tx) error {
+		for _, lockID := range lockOrder {
+			row := tx.QueryRow(ctx, `SELECT AcquireLock($1, $2)`, lockID, int(ttl.Seconds()))
+			if err := row.Scan(&expires); err != nil {
+				return err
+			}
+			if expires.Before(thePast) {
+				return ErrAlreadyLocked
+			}
+		}
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	logging.FromContext(ctx).Debugf("Acquired locks %v", lockOrder)
+	return makeMultiUnlockFn(ctx, db, lockOrder, expires), nil
+}
+
+func makeMultiUnlockFn(ctx context.Context, db *DB, lockIDs []string, expires time.Time) UnlockFn {
+	return func() error {
+		return db.InTx(ctx, pgx.Serializable, func(tx pgx.Tx) error {
+			for i := len(lockIDs) - 1; i >= 0; i-- {
+				lockID := lockIDs[i]
+				row := tx.QueryRow(ctx, `SELECT ReleaseLock($1, $2)`, lockID, expires)
+				var released bool
+				if err := row.Scan(&released); err != nil {
+					return err
+				}
+				if !released {
+					return fmt.Errorf("cannot delete lock %q that no longer belongs to you; it likely expired and was taken by another process", lockID)
+				}
+				logging.FromContext(ctx).Debugf("Released lock %q", lockID)
+			}
+			return nil
+		})
+	}
+}
+
 // Lock acquires lock with given name that times out after ttl. Returns an UnlockFn that can be used to unlock the lock. ErrAlreadyLocked will be returned if there is already a lock in use.
 func (db *DB) Lock(ctx context.Context, lockID string, ttl time.Duration) (UnlockFn, error) {
 	var expires time.Time
diff --git a/internal/export/config.go b/internal/export/config.go
index 5b39669..1ca0ca8 100644
--- a/internal/export/config.go
+++ b/internal/export/config.go
@@ -42,15 +42,16 @@ type Config struct {
 	Storage               storage.Config
 	ObservabilityExporter observability.Config
 
-	Port           string        `env:"PORT, default=8080"`
-	CreateTimeout  time.Duration `env:"CREATE_BATCHES_TIMEOUT, default=5m"`
-	WorkerTimeout  time.Duration `env:"WORKER_TIMEOUT, default=5m"`
-	MinRecords     int           `env:"EXPORT_FILE_MIN_RECORDS, default=1000"`
-	PaddingRange   int           `env:"EXPORT_FILE_PADDING_RANGE, default=100"`
-	MaxRecords     int           `env:"EXPORT_FILE_MAX_RECORDS, default=30000"`
-	TruncateWindow time.Duration `env:"TRUNCATE_WINDOW, default=1h"`
-	MinWindowAge   time.Duration `env:"MIN_WINDOW_AGE, default=2h"`
-	TTL            time.Duration `env:"CLEANUP_TTL, default=336h"`
+	Port               string        `env:"PORT, default=8080"`
+	CreateTimeout      time.Duration `env:"CREATE_BATCHES_TIMEOUT, default=5m"`
+	WorkerTimeout      time.Duration `env:"WORKER_TIMEOUT, default=5m"`
+	MinRecords         int           `env:"EXPORT_FILE_MIN_RECORDS, default=1000"`
+	PaddingRange       int           `env:"EXPORT_FILE_PADDING_RANGE, default=100"`
+	MaxRecords         int           `env:"EXPORT_FILE_MAX_RECORDS, default=30000"`
+	MaxInsertBatchSize int           `env:"MAX_INSERT_BATCH_SIZE, default=100"`
+	TruncateWindow     time.Duration `env:"TRUNCATE_WINDOW, default=1h"`
+	MinWindowAge       time.Duration `env:"MIN_WINDOW_AGE, default=2h"`
+	TTL                time.Duration `env:"CLEANUP_TTL, default=336h"`
 }
 
 func (c *Config) BlobstoreConfig() *storage.Config {
diff --git a/internal/export/database/export.go b/internal/export/database/export.go
index 4c83fe4..9be2cdc 100644
--- a/internal/export/database/export.go
+++ b/internal/export/database/export.go
@@ -522,6 +522,9 @@ func (db *ExportDB) LeaseBatch(ctx context.Context, ttl time.Duration, batchMaxC
 	var openBatchIDs []int64
 
 	if err := db.db.InTx(ctx, pgx.ReadCommitted, func(tx pgx.Tx) error {
+		// batches are ordered by end time. We try to fill the "oldest"
+		// batches first so that padding written will be covered by
+		// "newer" export batches.
 		rows, err := tx.Query(ctx, `
 			SELECT
 				batch_id
@@ -535,6 +538,8 @@ func (db *ExportDB) LeaseBatch(ctx context.Context, ttl time.Duration, batchMaxC
 				)
 			AND
 				end_timestamp < $3
+			ORDER BY
+				end_timestamp ASC
 			LIMIT 100
 		`, model.ExportBatchOpen, model.ExportBatchPending, batchMaxCloseTime)
 		if err != nil {
diff --git a/internal/export/worker.go b/internal/export/worker.go
index f507dc8..8bb1fee 100644
--- a/internal/export/worker.go
+++ b/internal/export/worker.go
@@ -42,6 +42,9 @@ import (
 const (
 	filenameSuffix       = ".zip"
 	blobOperationTimeout = 50 * time.Second
+
+	travelerLockID       = "TRAVELERS"
+	exportAppPackageName = "export-generated"
 )
 
 // handleDoWork is a handler to iterate the rows of ExportBatch, and creates
@@ -76,6 +79,32 @@ func (s *Server) handleDoWork(ctx context.Context) http.HandlerFunc {
 				return
 			}
 
+			// Obtain the necessary locks for this export batch. Ensure that only
+			// one export worker is operating over a region at a time.
+			//
+			// In the event that more than one exportconfig covers the same region (and travelers), it
+			// is important that only one export worker be allowed to generate keys for that region.
+			// An exclusive lock is taken before processing the batch over the input regions.
+			//
+			// In the export lease selection, we attempt to order export batch filling such that
+			// earlier batches are filled before later batches. This helps to reduce the possibility
+			// of non-overlapping generated data.
+			locks := make([]string, len(batch.EffectiveInputRegions()))
+			copy(locks, batch.EffectiveInputRegions())
+			if batch.IncludeTravelers {
+				locks = append(locks, travelerLockID)
+			}
+			unlockFn, err := db.MultiLock(ctx, locks, s.config.WorkerTimeout)
+			if err != nil {
+				logger.Errorw("unable to obtain region locks", "config", batch.ConfigID, "batch", batch.BatchID, "regions", locks, "error", err)
+				continue
+			}
+			unlock := func() {
+				if err := unlockFn(); err != nil {
+					logger.Errorw("failed to unlock region locks", batch.ConfigID, "batch", batch.BatchID, "regions", locks, "error", err)
+				}
+			}
+
 			// We re-write the index file for empty batches for self-healing so that the
 			// index file reflects the ExportFile table in database. However, if a
 			// single worker processes a number of empty batches quickly, we want to
@@ -88,10 +117,13 @@ func (s *Server) handleDoWork(ctx context.Context) http.HandlerFunc {
 				indexWrittenForConfig[batch.ConfigID] = struct{}{}
 			}
 
+			// Ensure that the locks are released on either success or failure path.
 			if err = s.exportBatch(ctx, batch, emitIndexForEmptyBatch); err != nil {
 				logger.Errorf("Failed to create files for batch: %v.", err)
+				unlock()
 				continue
 			}
+			unlock()
 
 			fmt.Fprintf(w, "Batch %d marked completed. \n", batch.BatchID)
 		}
@@ -107,25 +139,10 @@ func (g *group) Length() int {
 	return len(g.exposures) + len(g.revised)
 }
 
-func (s *Server) exportBatch(ctx context.Context, eb *model.ExportBatch, emitIndexForEmptyBatch bool) error {
+func (s *Server) batchExposures(ctx context.Context, criteria publishdatabase.IterateExposuresCriteria, outputRegion string) ([]*group, error) {
 	logger := logging.FromContext(ctx)
 	db := s.env.Database()
 
-	logger.Infof("Processing export batch %d (root: %q, region: %s), max records per file %d", eb.BatchID, eb.FilenameRoot, eb.OutputRegion, s.config.MaxRecords)
-
-	// Criteria starts w/ non-revised keys.
-	// Will be changed later to grab the revised keys.
-	criteria := publishdatabase.IterateExposuresCriteria{
-		SinceTimestamp:      eb.StartTimestamp,
-		UntilTimestamp:      eb.EndTimestamp,
-		IncludeRegions:      eb.EffectiveInputRegions(),
-		IncludeTravelers:    eb.IncludeTravelers, // Travelers are included from "any" region.
-		OnlyNonTravelers:    eb.OnlyNonTravelers,
-		ExcludeRegions:      eb.ExcludeRegions,
-		OnlyLocalProvenance: false, // include federated ids
-		OnlyRevisedKeys:     false,
-	}
-
 	// Build up groups of exposures in memory. We need to use memory so we can
 	// determine the total number of groups (which is embedded in each export
 	// file). This technique avoids SELECT COUNT which would lock the database
@@ -136,11 +153,17 @@ func (s *Server) exportBatch(ctx context.Context, eb *model.ExportBatch, emitInd
 	droppedKeys := 0
 
 	publishDB := publishdatabase.New(db)
+
+	maxCreatedAt := time.Time{}
 	_, err := publishDB.IterateExposures(ctx, criteria, func(exp *publishmodel.Exposure) error {
 		if len(exp.ExposureKey) != verifyapi.KeyLength {
 			droppedKeys++
 			return nil
 		}
+		// see if assigned time for generated data should be moved up.
+		if exp.CreatedAt.After(maxCreatedAt) {
+			maxCreatedAt = exp.CreatedAt
+		}
 		nextGroup.exposures = append(nextGroup.exposures, exp)
 		totalNewKeys++
 		if nextGroup.Length() == s.config.MaxRecords {
@@ -150,7 +173,7 @@ func (s *Server) exportBatch(ctx context.Context, eb *model.ExportBatch, emitInd
 		return nil
 	})
 	if err != nil {
-		return fmt.Errorf("iterating exposures: %w", err)
+		return nil, fmt.Errorf("iterating exposures: %w", err)
 	}
 
 	// go get the revised keys.
@@ -169,7 +192,7 @@ func (s *Server) exportBatch(ctx context.Context, eb *model.ExportBatch, emitInd
 		return nil
 	})
 	if err != nil {
-		return fmt.Errorf("iterating revised exposures: %w", err)
+		return nil, fmt.Errorf("iterating revised exposures: %w", err)
 	}
 
 	if droppedKeys > 0 {
@@ -185,20 +208,69 @@ func (s *Server) exportBatch(ctx context.Context, eb *model.ExportBatch, emitInd
 	}
 
 	if len(groups) == 0 {
-		logger.Infof("No records for export batch %d", eb.BatchID)
+		logger.Infof("No records for export batch")
 	} else {
 
 		lastGroup := groups[len(groups)-1]
-		lastGroup.exposures, err = ensureMinNumExposures(lastGroup.exposures, eb.OutputRegion, s.config.MinRecords, s.config.PaddingRange)
+		var generated []*publishmodel.Exposure
+		lastGroup.exposures, generated, err = ensureMinNumExposures(lastGroup.exposures, outputRegion, s.config.MinRecords, s.config.PaddingRange, maxCreatedAt)
 		if err != nil {
-			return fmt.Errorf("ensureMinNumExposures: %w", err)
+			return nil, fmt.Errorf("ensureMinNumExposures: %w", err)
 		}
-		lastGroup.revised, err = ensureMinNumExposures(lastGroup.revised, eb.OutputRegion, s.config.MinRecords, s.config.PaddingRange)
-		if err != nil {
-			return fmt.Errorf("ensureMinNumExposures: %w", err)
+		// padding revised keys doesn't provide any useful protection as one can work backwords and figure out which
+		// keys appeared as primary keys in a previous export.
+
+		if length := len(generated); length > 0 {
+			// we generated some data in order to pad out this export. This data needs to be persisted.
+			insertRequest := &publishdatabase.InsertAndReviseExposuresRequest{
+				RequireToken: false,
+				SkipRevions:  true,
+			}
+			// Insert the generated data in batches.
+			for i := 0; i < length; i = i + s.config.MaxInsertBatchSize {
+				upper := i + s.config.MaxInsertBatchSize
+				if upper > length {
+					upper = length
+				}
+				insertRequest.Incoming = generated[i:upper]
+				insertResponse, err := publishDB.InsertAndReviseExposures(ctx, insertRequest)
+				if err != nil {
+					// If this fails, the batch will be retried.
+					return nil, fmt.Errorf("writing generated data, publishDB.InsertAndReviseExposures: %w", err)
+				}
+				logger.Debugw("persisting generated keys", "num", insertResponse.Inserted)
+				i = i + upper
+			}
 		}
 	}
 
+	return groups, nil
+}
+
+func (s *Server) exportBatch(ctx context.Context, eb *model.ExportBatch, emitIndexForEmptyBatch bool) error {
+	logger := logging.FromContext(ctx)
+	db := s.env.Database()
+
+	logger.Infof("Processing export batch %d (root: %q, region: %s), max records per file %d", eb.BatchID, eb.FilenameRoot, eb.OutputRegion, s.config.MaxRecords)
+
+	// Criteria starts w/ non-revised keys.
+	// Will be changed later to grab the revised keys.
+	criteria := publishdatabase.IterateExposuresCriteria{
+		SinceTimestamp:      eb.StartTimestamp,
+		UntilTimestamp:      eb.EndTimestamp,
+		IncludeRegions:      eb.EffectiveInputRegions(),
+		IncludeTravelers:    eb.IncludeTravelers, // Travelers are included from "any" region.
+		OnlyNonTravelers:    eb.OnlyNonTravelers,
+		ExcludeRegions:      eb.ExcludeRegions,
+		OnlyLocalProvenance: false, // include federated ids
+		OnlyRevisedKeys:     false,
+	}
+
+	groups, err := s.batchExposures(ctx, criteria, eb.OutputRegion)
+	if err != nil {
+		return fmt.Errorf("reading exposures for batch: %w", err)
+	}
+
 	exportDB := exportdatabase.New(db)
 	// Load the non-expired signature infos associated with this export batch.
 	sigInfos, err := exportDB.LookupSignatureInfos(ctx, eb.SignatureInfoIDs, time.Now())
@@ -378,14 +450,16 @@ func randomInt(min, max int) (int, error) {
 	return int(n.Int64()) + min, nil
 }
 
-func ensureMinNumExposures(exposures []*publishmodel.Exposure, region string, minLength, jitter int) ([]*publishmodel.Exposure, error) {
-	if len(exposures) == 0 {
-		return exposures, nil
-	}
-
+func ensureMinNumExposures(exposures []*publishmodel.Exposure, region string, minLength, jitter int, createdAt time.Time) ([]*publishmodel.Exposure, []*publishmodel.Exposure, error) {
 	extra, _ := randomInt(0, jitter)
 	target := minLength + extra
+
+	if l := len(exposures); l == 0 || l >= target {
+		return exposures, make([]*publishmodel.Exposure, 0), nil
+	}
+
 	sourceLen := len(exposures) - 1
+	generated := make([]*publishmodel.Exposure, 0, target-len(exposures))
 
 	for len(exposures) < target {
 		// Pieces needed are
@@ -394,29 +468,35 @@ func ensureMinNumExposures(exposures []*publishmodel.Exposure, region string, mi
 		eKey := make([]byte, verifyapi.KeyLength)
 		_, err := rand.Read(eKey)
 		if err != nil {
-			return nil, fmt.Errorf("rand.Read: %w", err)
+			return nil, nil, fmt.Errorf("rand.Read: %w", err)
 		}
 
 		// Copy timing and report data from a key.
 		fromIdx, err := randomInt(0, sourceLen)
 		if err != nil {
-			return nil, fmt.Errorf("randomInt: %w", err)
+			return nil, nil, fmt.Errorf("randomInt: %w", err)
 		}
 
 		ek := &publishmodel.Exposure{
 			ExposureKey:                  eKey,
 			TransmissionRisk:             exposures[fromIdx].TransmissionRisk,
-			Regions:                      []string{region},
+			AppPackageName:               exportAppPackageName,
+			Regions:                      exposures[fromIdx].Regions,
+			Traveler:                     exposures[fromIdx].Traveler,
 			IntervalNumber:               exposures[fromIdx].IntervalNumber,
 			IntervalCount:                exposures[fromIdx].IntervalCount,
+			CreatedAt:                    createdAt,
+			LocalProvenance:              true,
 			ReportType:                   exposures[fromIdx].ReportType,
 			DaysSinceSymptomOnset:        exposures[fromIdx].DaysSinceSymptomOnset,
+			RevisedAt:                    exposures[fromIdx].RevisedAt,
 			RevisedReportType:            exposures[fromIdx].RevisedReportType,
 			RevisedDaysSinceSymptomOnset: exposures[fromIdx].RevisedDaysSinceSymptomOnset,
 			// The rest of the publishmodel.Exposure fields are not used in the export file.
 		}
+		generated = append(generated, ek)
 		exposures = append(exposures, ek)
 	}
 
-	return exposures, nil
+	return exposures, generated, nil
 }
diff --git a/internal/keyrotation/rotate.go b/internal/keyrotation/rotate.go
index 80df520..24db570 100644
--- a/internal/keyrotation/rotate.go
+++ b/internal/keyrotation/rotate.go
@@ -86,7 +86,7 @@ func (s *Server) doRotate(ctx context.Context) error {
 		previousCreated = allowed[0].CreatedAt
 	}
 
-	var result error
+	var result *multierror.Error
 	deleted := 0
 	for _, key := range allowed {
 		if did, err := s.maybeDeleteKey(ctx, key, effectiveID, previousCreated); err != nil {
@@ -100,7 +100,7 @@ func (s *Server) doRotate(ctx context.Context) error {
 		logger.Infof("Deleted %d old revision keys.", deleted)
 		metricsMiddleware.RecordRevisionKeyDeletion(ctx, deleted)
 	}
-	return result
+	return result.ErrorOrNil()
 }
 
 func (s *Server) maybeDeleteKey(
diff --git a/internal/pb/federation/federation.pb.go b/internal/pb/federation/federation.pb.go
index 7f23126..006a520 100644
--- a/internal/pb/federation/federation.pb.go
+++ b/internal/pb/federation/federation.pb.go
@@ -116,8 +116,7 @@ type FederationFetchRequest struct {
 	MaxExposureKeys uint32 `protobuf:"varint,5,opt,name=maxExposureKeys,proto3" json:"maxExposureKeys,omitempty"`
 	// region, includeTravelers, onlyTravelers must be stable to send a fetchToken.
 	// initial query should send an empty fetch state token.
-	State            *FetchState `protobuf:"bytes,6,opt,name=state,proto3" json:"state,omitempty"`
-	OnlyNonTravelers bool        `protobuf:"varint,7,opt,name=onlyNonTravelers,proto3" json:"onlyNonTravelers,omitempty"`
+	State *FetchState `protobuf:"bytes,6,opt,name=state,proto3" json:"state,omitempty"`
 }
 
 func (x *FederationFetchRequest) Reset() {
diff --git a/internal/publish/config.go b/internal/publish/config.go
index bebf8ad..559de6e 100644
--- a/internal/publish/config.go
+++ b/internal/publish/config.go
@@ -81,6 +81,9 @@ type Config struct {
 	// Should only be set if a server is being operated in a single region.
 	DefaultRegion string `env:"DEFAULT_REGION"`
 
+	// Feature flags - eventually these are removed as features become default behavior
+	FailOnCertificateAudienceMismatch bool `env:"FEATURE_FAIL_ON_CERTIFICATE_AUDIENCE_MISMATCH, default=false"`
+
 	// Flags for local development and testing. This will cause still valid keys
 	// to not be embargoed.
 	// Normally "still valid" keys can be accepted, but are embargoed.
diff --git a/internal/publish/database/exposure.go b/internal/publish/database/exposure.go
index 08b32c2..c7dd181 100644
--- a/internal/publish/database/exposure.go
+++ b/internal/publish/database/exposure.go
@@ -273,8 +273,7 @@ func generateExposureQuery(criteria IterateExposuresCriteria) (string, []interfa
 func (db *PublishDB) ReadExposures(ctx context.Context, tx pgx.Tx, b64keys []string) (map[string]*model.Exposure, error) {
 	exposures := make(map[string]*model.Exposure)
 
-	if err := db.db.InTx(ctx, pgx.ReadCommitted, func(tx pgx.Tx) error {
-		rows, err := tx.Query(ctx, `
+	rows, err := tx.Query(ctx, `
 			SELECT
 				exposure_key, transmission_risk, app_package_name, regions, traveler,
 				interval_number, interval_count, created_at, local_provenance, sync_id,
@@ -286,47 +285,42 @@ func (db *PublishDB) ReadExposures(ctx context.Context, tx pgx.Tx, b64keys []str
 			WHERE exposure_key = ANY($1)
 			FOR UPDATE
 		`, b64keys)
-		if err != nil {
-			return fmt.Errorf("failed to list: %w", err)
-		}
-		defer rows.Close()
-
-		for rows.Next() {
-			if err := rows.Err(); err != nil {
-				return fmt.Errorf("failed to iterate: %w", err)
-			}
+	if err != nil {
+		return nil, fmt.Errorf("failed to list: %w", err)
+	}
+	defer rows.Close()
 
-			var encodedKey string
-			var syncID sql.NullInt64
-
-			var exposure model.Exposure
-			if err := rows.Scan(
-				&encodedKey, &exposure.TransmissionRisk, &exposure.AppPackageName,
-				&exposure.Regions, &exposure.Traveler, &exposure.IntervalNumber, &exposure.IntervalCount,
-				&exposure.CreatedAt, &exposure.LocalProvenance, &syncID,
-				&exposure.HealthAuthorityID, &exposure.ReportType, &exposure.DaysSinceSymptomOnset,
-				&exposure.RevisedReportType, &exposure.RevisedAt, &exposure.RevisedDaysSinceSymptomOnset,
-				&exposure.RevisedTransmissionRisk, &exposure.ExportImportID,
-			); err != nil {
-				return fmt.Errorf("failed to parse: %w", err)
-			}
+	for rows.Next() {
+		if err := rows.Err(); err != nil {
+			return nil, fmt.Errorf("failed to iterate: %w", err)
+		}
 
-			// Base64 decode the exposure key
-			exposure.ExposureKey, err = decodeExposureKey(encodedKey)
-			if err != nil {
-				return fmt.Errorf("failed to decode key: %w", err)
-			}
-			// Optionally set all of the nullable columns.
-			if syncID.Valid {
-				exposure.FederationSyncID = syncID.Int64
-			}
+		var encodedKey string
+		var syncID sql.NullInt64
+
+		var exposure model.Exposure
+		if err := rows.Scan(
+			&encodedKey, &exposure.TransmissionRisk, &exposure.AppPackageName,
+			&exposure.Regions, &exposure.Traveler, &exposure.IntervalNumber, &exposure.IntervalCount,
+			&exposure.CreatedAt, &exposure.LocalProvenance, &syncID,
+			&exposure.HealthAuthorityID, &exposure.ReportType, &exposure.DaysSinceSymptomOnset,
+			&exposure.RevisedReportType, &exposure.RevisedAt, &exposure.RevisedDaysSinceSymptomOnset,
+			&exposure.RevisedTransmissionRisk, &exposure.ExportImportID,
+		); err != nil {
+			return nil, fmt.Errorf("failed to parse: %w", err)
+		}
 
-			exposures[exposure.ExposureKeyBase64()] = &exposure
+		// Base64 decode the exposure key
+		exposure.ExposureKey, err = decodeExposureKey(encodedKey)
+		if err != nil {
+			return nil, fmt.Errorf("failed to decode key: %w", err)
+		}
+		// Optionally set all of the nullable columns.
+		if syncID.Valid {
+			exposure.FederationSyncID = syncID.Int64
 		}
 
-		return nil
-	}); err != nil {
-		return nil, fmt.Errorf("read exposures: %w", err)
+		exposures[exposure.ExposureKeyBase64()] = &exposure
 	}
 
 	return exposures, nil
diff --git a/internal/publish/publish.go b/internal/publish/publish.go
index 688ac86..7eb1c5a 100644
--- a/internal/publish/publish.go
+++ b/internal/publish/publish.go
@@ -277,7 +277,7 @@ func (h *PublishHandler) process(ctx context.Context, data *verifyapi.Publish, b
 	}
 
 	// Perform health authority certificate verification.
-	verifiedClaims, err := h.verifier.VerifyDiagnosisCertificate(ctx, appConfig, data)
+	verifiedClaims, err := h.verifier.VerifyDiagnosisCertificate(ctx, appConfig, data, h.config.FailOnCertificateAudienceMismatch)
 	if err != nil {
 		if appConfig.BypassHealthAuthorityVerification {
 			logger.Warnf("bypassing health authority certificate verification health authority: %v", appConfig.AppPackageName)
diff --git a/internal/setup/setup.go b/internal/setup/setup.go
index f623ca7..bacc6a2 100644
--- a/internal/setup/setup.go
+++ b/internal/setup/setup.go
@@ -172,7 +172,7 @@ func SetupWith(ctx context.Context, config interface{}, l envconfig.Lookuper) (*
 		logger.Info("configuring observability exporter")
 
 		oeConfig := provider.ObservabilityExporterConfig()
-		oe, err := observability.NewFromEnv(ctx, oeConfig)
+		oe, err := observability.NewFromEnv(oeConfig)
 		if err != nil {
 			return nil, fmt.Errorf("unable to create observability provider: %v", err)
 		}
diff --git a/internal/verification/database/health_authority_db.go b/internal/verification/database/health_authority_db.go
index 261434a..186db22 100644
--- a/internal/verification/database/health_authority_db.go
+++ b/internal/verification/database/health_authority_db.go
@@ -27,6 +27,10 @@ import (
 	pgx "github.com/jackc/pgx/v4"
 )
 
+var (
+	ErrHealthAuthorityNotFound = errors.New("health authority not found")
+)
+
 // HealthAuthorityDB allows for opreations against authorized health authorities
 // for diagnosis signature verification.
 type HealthAuthorityDB struct {
@@ -141,10 +145,13 @@ func (db *HealthAuthorityDB) GetHealthAuthority(ctx context.Context, issuer stri
 		var err error
 		ha, err = scanOneHealthAuthority(row)
 		if err != nil {
-			return fmt.Errorf("failed to parse: %w", err)
+			return err
 		}
 		return nil
 	}); err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return nil, ErrHealthAuthorityNotFound
+		}
 		return nil, fmt.Errorf("get health authority: %w", err)
 	}
 
diff --git a/internal/verification/phaverify.go b/internal/verification/phaverify.go
index c5f93da..e381338 100644
--- a/internal/verification/phaverify.go
+++ b/internal/verification/phaverify.go
@@ -20,6 +20,7 @@ package verification
 import (
 	"context"
 	"crypto/hmac"
+	"errors"
 	"fmt"
 
 	aamodel "github.com/google/exposure-notifications-server/internal/authorizedapp/model"
@@ -28,6 +29,7 @@ import (
 	verifyapi "github.com/google/exposure-notifications-server/pkg/api/v1"
 	"github.com/google/exposure-notifications-server/pkg/base64util"
 	"github.com/google/exposure-notifications-server/pkg/cache"
+	"github.com/google/exposure-notifications-server/pkg/logging"
 	utils "github.com/google/exposure-notifications-server/pkg/verification"
 
 	"github.com/dgrijalva/jwt-go"
@@ -59,7 +61,8 @@ type VerifiedClaims struct {
 // VerifyDiagnosisCertificate accepts a publish request (from which is extracts the JWT),
 // fully verifies the JWT and signture against what the passed in authorrized app is allowed
 // to use. Returns any transmission risk overrides if they are present.
-func (v *Verifier) VerifyDiagnosisCertificate(ctx context.Context, authApp *aamodel.AuthorizedApp, publish *verifyapi.Publish) (*VerifiedClaims, error) {
+func (v *Verifier) VerifyDiagnosisCertificate(ctx context.Context, authApp *aamodel.AuthorizedApp, publish *verifyapi.Publish, enforceAudienceMatch bool) (*VerifiedClaims, error) {
+	logger := logging.FromContext(ctx)
 	// These get assigned during the ParseWithClaims closure.
 	var healthAuthorityID int64
 	var claims *verifyapi.VerificationClaims
@@ -84,6 +87,11 @@ func (v *Verifier) VerifyDiagnosisCertificate(ctx context.Context, authApp *aamo
 		lookup := func() (interface{}, error) {
 			// Based on issuer, load the key versions.
 			ha, err := v.db.GetHealthAuthority(ctx, claims.Issuer)
+			// Special case not found so that we can cache it.
+			if errors.Is(err, database.ErrHealthAuthorityNotFound) {
+				logger.Warnw("requested issuer not found", "iss", claims.Issuer)
+				return nil, nil
+			}
 			if err != nil {
 				return nil, fmt.Errorf("error looking up issuer: %v : %w", claims.Issuer, err)
 			}
@@ -94,8 +102,22 @@ func (v *Verifier) VerifyDiagnosisCertificate(ctx context.Context, authApp *aamo
 			return nil, err
 		}
 
+		if cacheVal == nil {
+			return nil, fmt.Errorf("issuer not found: %v", claims.Issuer)
+		}
+
 		ha := cacheVal.(*model.HealthAuthority)
 
+		// Advisory check the aud.
+		if claims.Audience != ha.Audience {
+			// TODO(mikehelmick) - clean up feature flag casing.
+			logger.Errorw("certifice audience mismatch - will be a failure in the next release", "claims.Aud", claims.Audience, "allowed", ha.Audience, "iss", ha.Issuer)
+			if enforceAudienceMatch {
+				// This flag guarding will be removed in a future release.
+				return nil, fmt.Errorf("audience mismatch for issuer: %v", ha.Issuer)
+			}
+		}
+
 		// Find a key version.
 		for _, hak := range ha.Keys {
 			// Key version matches and the key is valid based on the current time.
diff --git a/pkg/observability/observability.go b/pkg/observability/observability.go
index 6497c57..d3be329 100644
--- a/pkg/observability/observability.go
+++ b/pkg/observability/observability.go
@@ -65,7 +65,11 @@ type Exporter interface {
 
 // NewFromEnv returns the observability exporter given the provided configuration, or an error
 // if it failed to be created.
-func NewFromEnv(ctx context.Context, config *Config) (Exporter, error) {
+func NewFromEnv(config *Config) (Exporter, error) {
+	// Create a separate ctx.
+	// The main ctx will be canceled when the server is shutting down. Sharing
+	// the main ctx prevent the last batch of the metrics to be uploaded.
+	ctx := context.Background()
 	switch config.ExporterType {
 	case ExporterNoop:
 		return NewNoop(ctx)
diff --git a/pkg/server/server.go b/pkg/server/server.go
index 72ea365..0e6722c 100644
--- a/pkg/server/server.go
+++ b/pkg/server/server.go
@@ -62,6 +62,22 @@ func New(port string) (*Server, error) {
 	}, nil
 }
 
+// NewFromListener creates a new server on the given listener. This is useful if
+// you want to customize the listener type (e.g. udp or tcp) or bind network
+// more than `New` allows.
+func NewFromListener(listener net.Listener) (*Server, error) {
+	addr, ok := listener.Addr().(*net.TCPAddr)
+	if !ok {
+		return nil, fmt.Errorf("listener is not tcp")
+	}
+
+	return &Server{
+		ip:       addr.IP.String(),
+		port:     strconv.Itoa(addr.Port),
+		listener: listener,
+	}, nil
+}
+
 // ServeHTTP starts the server and blocks until the provided context is closed.
 // When the provided context is closed, the server is gracefully stopped with a
 // timeout of 5 seconds.
diff --git a/tools/export-analyzer/main.go b/tools/export-analyzer/main.go
index 4f8cac7..a256481 100644
--- a/tools/export-analyzer/main.go
+++ b/tools/export-analyzer/main.go
@@ -113,11 +113,7 @@ func checkExportFile(export *exportpb.TemporaryExposureKeyExport) error {
 	if err := checkKeys("revisedKeys", export.RevisedKeys, floor, ceiling); err != nil {
 		errors = multierror.Append(errors, err)
 	}
-
-	if len(errors.Errors) != 0 {
-		return errors
-	}
-	return nil
+	return errors.ErrorOrNil()
 }
 
 func checkKeys(kType string, keys []*exportpb.TemporaryExposureKey, floor, ceiling int32) error {


Test patch:
diff --git a/docker/presubmit-test.dockerfile b/docker/presubmit-test.dockerfile
index a3d3aa3..7e0d0ec 100644
--- a/docker/presubmit-test.dockerfile
+++ b/docker/presubmit-test.dockerfile
@@ -14,9 +14,6 @@
 
 # This image is used to run ./scripts/presubmit.sh on CI
 
-# Stable as of 10/22/2020
-FROM chromedp/headless-shell:86.0.4240.111 as headlessshell
-
 FROM golang:1.15.2
 
 # Install sudo
@@ -50,6 +47,12 @@ RUN wget -q https://dl.google.com/dl/cloudsdk/channels/rapid/google-cloud-sdk.ta
     gcloud components install alpha beta kubectl && \
     gcloud info | tee /workspace/gcloud-info.txt
 
+# Install Chrome - used for browser tests
+RUN wget -q -O - https://dl-ssl.google.com/linux/linux_signing_key.pub | apt-key add - && \
+  sh -c 'echo "deb [arch=amd64] http://dl.google.com/linux/chrome/deb/ stable main" >> /etc/apt/sources.list.d/google.list' && \
+  apt-get update && apt-get install -yqq google-chrome-stable
+
+
 #
 # BEGIN: DOCKER IN DOCKER SETUP
 #
@@ -101,12 +104,4 @@ RUN go get -u honnef.co/go/tools/cmd/staticcheck
 # GCP projects pool manager
 RUN go get -u sigs.k8s.io/boskos/cmd/boskosctl
 
-# Copy headless-shell and add it to PATH
-WORKDIR /headless-shell
-RUN mkdir -p /headless-shell
-COPY --from=headlessshell /headless-shell/headless-shell /headless-shell/headless-shell
-# "libnss3.so" under /usr/lib is required by headless-shell, also copy this
-COPY --from=headlessshell /usr/lib/* /usr/lib/
-ENV PATH=/headless-shell:/workspace:${PATH}
-
 ENTRYPOINT ["/bin/runner.sh"]
diff --git a/internal/database/lock_test.go b/internal/database/lock_test.go
index d9da0ea..68bb4be 100644
--- a/internal/database/lock_test.go
+++ b/internal/database/lock_test.go
@@ -83,3 +83,46 @@ func TestLock(t *testing.T) {
 		t.Fatalf("got %d rows from Lock table, wanted zero", count)
 	}
 }
+
+func TestMultiLock(t *testing.T) {
+	t.Parallel()
+
+	testDB := NewTestDatabase(t)
+	ctx := context.Background()
+
+	neededLocks := []string{"traveler", "US", "CA", "MX"}
+
+	unlock1, err := testDB.MultiLock(ctx, neededLocks, time.Minute)
+	if err != nil {
+		t.Fatalf("failed to obtain locks, %v, err: %v", neededLocks, err)
+	}
+
+	overlappingLocks := []string{"CA", "CH", "UK"}
+
+	if _, err := testDB.MultiLock(ctx, overlappingLocks, time.Minute); err == nil {
+		t.Fatalf("expected lock acquisition to fail, but didn't.")
+	} else if !errors.Is(err, ErrAlreadyLocked) {
+		t.Fatalf("wong error want: %v, got: %v", ErrAlreadyLocked, err)
+	}
+
+	nonoverlappingLocks := []string{"CH", "UK"}
+	unlock2, err := testDB.MultiLock(ctx, nonoverlappingLocks, time.Minute)
+	if err != nil {
+		t.Fatalf("failed to obtain locks, %v, err: %v", nonoverlappingLocks, err)
+	}
+
+	if err := unlock1(); err != nil {
+		t.Fatalf("failed to release locks: %v", err)
+	}
+
+	// should still fail, because there is still ovelap w/ the second lock.
+	if _, err := testDB.MultiLock(ctx, overlappingLocks, time.Minute); err == nil {
+		t.Fatalf("expected lock acquisition to fail, but didn't.")
+	} else if !errors.Is(err, ErrAlreadyLocked) {
+		t.Fatalf("wong error want: %v, got: %v", ErrAlreadyLocked, err)
+	}
+
+	if err := unlock2(); err != nil {
+		t.Fatalf("failed to release locks: %v", err)
+	}
+}
diff --git a/internal/export/worker_test.go b/internal/export/worker_test.go
index 9dcddeb..9565f72 100644
--- a/internal/export/worker_test.go
+++ b/internal/export/worker_test.go
@@ -15,11 +15,17 @@
 package export
 
 import (
+	"context"
 	"testing"
+	"time"
 
-	"github.com/google/exposure-notifications-server/internal/publish/model"
+	"github.com/google/exposure-notifications-server/internal/database"
+	publishdb "github.com/google/exposure-notifications-server/internal/publish/database"
+	publishmodel "github.com/google/exposure-notifications-server/internal/publish/model"
+	"github.com/google/exposure-notifications-server/internal/serverenv"
 	verifyapi "github.com/google/exposure-notifications-server/pkg/api/v1alpha1"
 	"github.com/google/exposure-notifications-server/pkg/util"
+	"github.com/google/go-cmp/cmp"
 	"google.golang.org/protobuf/proto"
 )
 
@@ -47,19 +53,22 @@ func TestRandomInt(t *testing.T) {
 }
 
 func TestDoNotPadZeroLength(t *testing.T) {
-	exposures := make([]*model.Exposure, 0)
-	exposures, err := ensureMinNumExposures(exposures, "US", 1000, 100)
+	exposures := make([]*publishmodel.Exposure, 0)
+	exposures, generated, err := ensureMinNumExposures(exposures, "US", 1000, 100, time.Now())
 	if err != nil {
 		t.Fatalf("unepected error: %v", err)
 	}
 	if len(exposures) != 0 {
 		t.Errorf("empty exposure list got padded, shouldn't have.")
 	}
+	if len(generated) != 0 {
+		t.Errorf("generated data returned, should be empty")
+	}
 }
 
 func TestEnsureMinExposures(t *testing.T) {
 	// Insert a few exposures - that will be used to base the interval information off of.
-	exposures := []*model.Exposure{
+	exposures := []*publishmodel.Exposure{
 		{
 			TransmissionRisk:      verifyapi.TransmissionRiskConfirmedStandard,
 			IntervalNumber:        123456,
@@ -99,13 +108,17 @@ func TestEnsureMinExposures(t *testing.T) {
 	m[123456+144] = 1
 
 	// pad the download.
-	exposures, err := ensureMinNumExposures(exposures, "US", numKeys, variance)
+	inputSize := len(exposures)
+	exposures, generated, err := ensureMinNumExposures(exposures, "US", numKeys, variance, time.Now())
 	if err != nil {
 		t.Fatalf("unexpected error: %v", err)
 	}
 	if len(exposures) < numKeys || len(exposures) > numKeys+variance {
 		t.Errorf("wrong number of exposures, want: >=%v and <=%v, got: %v", numKeys, numKeys+variance, len(exposures))
 	}
+	if l, exp := len(generated), numKeys-inputSize; l < exp {
+		t.Errorf("want keys >= %d, got: %d", exp, l)
+	}
 
 	for _, e := range exposures {
 		m[e.IntervalNumber] = m[e.IntervalNumber] + 1
@@ -116,3 +129,203 @@ func TestEnsureMinExposures(t *testing.T) {
 		}
 	}
 }
+
+func getKey(t *testing.T) []byte {
+	t.Helper()
+	eKey, err := util.RandomBytes(verifyapi.KeyLength)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return eKey
+}
+
+func TestBatchExposures(t *testing.T) {
+	t.Parallel()
+
+	testDB := database.NewTestDatabase(t)
+	testPublishDB := publishdb.New(testDB)
+	ctx := context.Background()
+
+	config := Config{
+		MinRecords:         5,
+		PaddingRange:       0,
+		TruncateWindow:     time.Hour,
+		MaxInsertBatchSize: 100,
+	}
+	server := Server{
+		config: &config,
+		env:    serverenv.New(ctx, serverenv.WithDatabase(testDB)),
+	}
+
+	// Using a 1 hour truncate window
+	// * one export batch lineage w/ 1 hour window
+	// * one export batch wiuth 4 hour window
+	baseTime := time.Date(2020, 10, 28, 1, 0, 0, 0, time.UTC).Truncate(time.Hour)
+	exposures := make([]*publishmodel.Exposure, 12)
+	for i := 0; i < 4; i++ {
+		// home country non-traveler
+		exposures[i*3] = &publishmodel.Exposure{
+			ExposureKey:     getKey(t),
+			Regions:         []string{"US"},
+			IntervalNumber:  100,
+			IntervalCount:   144,
+			CreatedAt:       baseTime.Add(time.Duration(i) * time.Hour),
+			LocalProvenance: true,
+			Traveler:        false,
+		}
+		// foreign country traveler
+		exposures[i*3+1] = &publishmodel.Exposure{
+			ExposureKey:     getKey(t),
+			Regions:         []string{"CA"},
+			IntervalNumber:  100,
+			IntervalCount:   144,
+			CreatedAt:       baseTime.Add(time.Duration(i) * time.Hour),
+			LocalProvenance: false,
+			Traveler:        true,
+		}
+		// foreign country non-traveler
+		exposures[i*3+2] = &publishmodel.Exposure{
+			ExposureKey:     getKey(t),
+			Regions:         []string{"CA"},
+			IntervalNumber:  100,
+			IntervalCount:   144,
+			CreatedAt:       baseTime.Add(time.Duration(i) * time.Hour),
+			LocalProvenance: false,
+			Traveler:        false,
+		}
+	}
+	if _, err := testPublishDB.InsertAndReviseExposures(ctx, &publishdb.InsertAndReviseExposuresRequest{
+		Incoming:     exposures,
+		RequireToken: true,
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	homePlusTraveler := make(map[string]struct{})
+	// Create the home country + traveler exports. 4, 1 per hour.
+	for i := 0; i < 4; i++ {
+		// Build the iteration criteria for the incremental batches.
+		criteria := publishdb.IterateExposuresCriteria{
+			SinceTimestamp:      baseTime.Add(time.Duration(i) * time.Hour),
+			UntilTimestamp:      baseTime.Add(time.Duration(i) * time.Hour).Add(time.Hour),
+			IncludeRegions:      []string{"US"},
+			IncludeTravelers:    true,
+			OnlyLocalProvenance: true,
+		}
+
+		groups, err := server.batchExposures(ctx, criteria, "US")
+		if err != nil {
+			t.Fatalf("failed to read exposures: %v", err)
+		}
+		if len(groups) == 0 {
+			t.Fatalf("export batch should have found some keys")
+		}
+
+		for _, group := range groups {
+			for _, exp := range group.exposures {
+				b64 := exp.ExposureKeyBase64()
+				if _, ok := homePlusTraveler[b64]; ok {
+					t.Fatalf("hourly batch included duplicate key")
+				}
+				homePlusTraveler[b64] = struct{}{}
+			}
+		}
+	}
+
+	foreignNonTraveler := make(map[string]struct{})
+	// Create the foreign, non traveler export.
+	for i := 0; i < 4; i++ {
+		// Build the iteration criteria for the incremental batches.
+		criteria := publishdb.IterateExposuresCriteria{
+			SinceTimestamp:      baseTime.Add(time.Duration(i) * time.Hour),
+			UntilTimestamp:      baseTime.Add(time.Duration(i) * time.Hour).Add(time.Hour),
+			IncludeRegions:      []string{"CA"}, // current list of foreign countries
+			OnlyNonTravelers:    true,
+			OnlyLocalProvenance: false,
+		}
+
+		groups, err := server.batchExposures(ctx, criteria, "REMOTE")
+		if err != nil {
+			t.Fatalf("failed to read exposures: %v", err)
+		}
+		if len(groups) == 0 {
+			t.Fatalf("export batch should have found some keys")
+		}
+
+		for _, group := range groups {
+			for _, exp := range group.exposures {
+				b64 := exp.ExposureKeyBase64()
+				if _, ok := foreignNonTraveler[b64]; ok {
+					t.Fatalf("hourly batch included duplicate key")
+				}
+				foreignNonTraveler[b64] = struct{}{}
+			}
+		}
+	}
+
+	// Run the 4 hour export for home+traveler
+	{
+		criteria := publishdb.IterateExposuresCriteria{
+			SinceTimestamp:      baseTime,
+			UntilTimestamp:      baseTime.Add(4 * time.Hour),
+			IncludeRegions:      []string{"US"},
+			IncludeTravelers:    true,
+			OnlyLocalProvenance: true,
+		}
+		groups, err := server.batchExposures(ctx, criteria, "US")
+		if err != nil {
+			t.Fatalf("failed to read exposures: %v", err)
+		}
+		if len(groups) == 0 {
+			t.Fatalf("export batch should have keys")
+		}
+
+		homePlusTravelerRollup := make(map[string]struct{})
+		for _, group := range groups {
+			for _, exp := range group.exposures {
+				b64 := exp.ExposureKeyBase64()
+				if _, ok := homePlusTravelerRollup[b64]; ok {
+					t.Fatalf("home rollup included duplicate key")
+				}
+				homePlusTravelerRollup[b64] = struct{}{}
+			}
+		}
+
+		if diff := cmp.Diff(homePlusTraveler, homePlusTravelerRollup); diff != "" {
+			t.Errorf("ReadExposures mismatch (-want, +got):\n%s", diff)
+		}
+	}
+
+	// Run the 4 hour foreign, non-traveler
+	{
+		criteria := publishdb.IterateExposuresCriteria{
+			SinceTimestamp:      baseTime,
+			UntilTimestamp:      baseTime.Add(4 * time.Hour),
+			IncludeRegions:      []string{"CA"}, // current list of foreign countries
+			OnlyNonTravelers:    true,
+			OnlyLocalProvenance: false,
+		}
+		groups, err := server.batchExposures(ctx, criteria, "REMOTE")
+		if err != nil {
+			t.Fatalf("failed to read exposures: %v", err)
+		}
+		if len(groups) == 0 {
+			t.Fatalf("export batch should have keys")
+		}
+
+		foreignNonTravelerRollup := make(map[string]struct{})
+		for _, group := range groups {
+			for _, exp := range group.exposures {
+				b64 := exp.ExposureKeyBase64()
+				if _, ok := foreignNonTravelerRollup[b64]; ok {
+					t.Fatalf("home rollup included duplicate key")
+				}
+				foreignNonTravelerRollup[b64] = struct{}{}
+			}
+		}
+
+		if diff := cmp.Diff(foreignNonTraveler, foreignNonTravelerRollup); diff != "" {
+			t.Errorf("ReadExposures mismatch (-want, +got):\n%s", diff)
+		}
+	}
+}
diff --git a/internal/verification/database/health_authority_db_test.go b/internal/verification/database/health_authority_db_test.go
index 8292fee..e813e42 100644
--- a/internal/verification/database/health_authority_db_test.go
+++ b/internal/verification/database/health_authority_db_test.go
@@ -16,6 +16,7 @@ package database
 
 import (
 	"context"
+	"errors"
 	"testing"
 	"time"
 
@@ -32,6 +33,22 @@ IBSEEHOdgpAynz0yrHpkWL6vxjNHxRdWcImZxPgL0NVHMdY4TlsL7qaxBQ==
 -----END PUBLIC KEY-----`
 )
 
+func TestMissingHealthAuthority(t *testing.T) {
+	t.Parallel()
+
+	testDB := database.NewTestDatabase(t)
+	haDB := New(testDB)
+	ctx := context.Background()
+
+	_, err := haDB.GetHealthAuthority(ctx, "does-not-exist")
+	if err == nil {
+		t.Fatal("missing error")
+	}
+	if !errors.Is(err, ErrHealthAuthorityNotFound) {
+		t.Fatalf("wrong error want: %v got: %v", ErrHealthAuthorityNotFound, err)
+	}
+}
+
 func TestAddRetrieveHealthAuthority(t *testing.T) {
 	t.Parallel()
 
diff --git a/internal/verification/phaverify_test.go b/internal/verification/phaverify_test.go
index f236a2a..37ae0a2 100644
--- a/internal/verification/phaverify_test.go
+++ b/internal/verification/phaverify_test.go
@@ -25,6 +25,7 @@ import (
 	"crypto/x509"
 	"encoding/base64"
 	"encoding/pem"
+	"fmt"
 	"strings"
 	"testing"
 	"time"
@@ -50,11 +51,30 @@ func TestVerifyCertificate(t *testing.T) {
 		Warp             time.Duration
 		MacAdjustment    string
 		MacKeyAdjustment string
+		ChangeIssuer     string
+		ChangeAudience   string
+		EnforceAudience  bool // TODO(mikehelmick) Remove when removing feature flag
 		Error            string
 	}{
 		{
 			Name: "happy path, valid cert",
 		},
+		{
+			Name:         "bad_issuer",
+			ChangeIssuer: "foo",
+			Error:        "issuer not found",
+		},
+		{
+			Name:            "bad_audience_off",
+			ChangeAudience:  "bar",
+			EnforceAudience: false,
+		},
+		{
+			Name:            "bad_audience",
+			ChangeAudience:  "bar",
+			Error:           "audience mismatch for issuer",
+			EnforceAudience: true,
+		},
 		{
 			Name:  "past",
 			Warp:  -1 * time.Hour,
@@ -77,6 +97,11 @@ func TestVerifyCertificate(t *testing.T) {
 		},
 	}
 
+	// Set up database. Create HealthAuthority + HAKey for the test.
+	testDB := coredb.NewTestDatabase(t)
+	ctx := context.Background()
+	haDB := database.New(testDB)
+
 	for iteration := 0; iteration < 2; iteration++ {
 		for version := 0; version <= 1; version++ {
 			for _, tc := range cases {
@@ -109,14 +134,11 @@ func TestVerifyCertificate(t *testing.T) {
 					pemEncodedPub := pem.EncodeToMemory(&pem.Block{Type: "PUBLIC KEY", Bytes: x509EncodedPub})
 					pemPublicKey := string(pemEncodedPub)
 
-					// Set up database. Create HealthAuthority + HAKey for the test.
-					testDB := coredb.NewTestDatabase(t)
-					ctx := context.Background()
-					haDB := database.New(testDB)
-
+					issuer := fmt.Sprintf("issuer-%s-%s-%s", vname, mod, t.Name())
+					audience := fmt.Sprintf("aud-%s-%s-%s", vname, mod, t.Name())
 					healthAuthority := model.HealthAuthority{
-						Issuer:   "doh.my.gov",
-						Audience: "exposure-notifications-server",
+						Issuer:   issuer,
+						Audience: audience,
 						Name:     "Very Real Health Authority",
 					}
 					if err := haDB.AddHealthAuthority(ctx, &healthAuthority); err != nil {
@@ -186,11 +208,18 @@ func TestVerifyCertificate(t *testing.T) {
 						hmac = allHMACs[1]
 					}
 
+					if tc.ChangeIssuer != "" {
+						issuer = tc.ChangeIssuer
+					}
+					if tc.ChangeAudience != "" {
+						audience = tc.ChangeAudience
+					}
+
 					var claims jwt.Claims
 					if version == 0 {
 						v1alpha1claims := v1alpha1.NewVerificationClaims()
-						v1alpha1claims.Audience = "exposure-notifications-server"
-						v1alpha1claims.Issuer = "doh.my.gov"
+						v1alpha1claims.Audience = audience
+						v1alpha1claims.Issuer = issuer
 						v1alpha1claims.IssuedAt = time.Now().Add(tc.Warp).Unix()
 						v1alpha1claims.ExpiresAt = time.Now().Add(tc.Warp).Add(5 * time.Minute).Unix()
 						v1alpha1claims.SignedMAC = tc.MacAdjustment + base64.StdEncoding.EncodeToString(hmac) // would be generated on the client and passed through.
@@ -200,8 +229,8 @@ func TestVerifyCertificate(t *testing.T) {
 						claims = v1alpha1claims
 					} else {
 						v1claims := verifyapi.NewVerificationClaims()
-						v1claims.Audience = "exposure-notifications-server"
-						v1claims.Issuer = "doh.my.gov"
+						v1claims.Audience = audience
+						v1claims.Issuer = issuer
 						v1claims.IssuedAt = time.Now().Add(tc.Warp).Unix()
 						v1claims.ExpiresAt = time.Now().Add(tc.Warp).Add(5 * time.Minute).Unix()
 						v1claims.SignedMAC = tc.MacAdjustment + base64.StdEncoding.EncodeToString(hmac)
@@ -226,9 +255,11 @@ func TestVerifyCertificate(t *testing.T) {
 					if err != nil {
 						t.Fatal(err)
 					}
-					verifiedClaims, err := verifier.VerifyDiagnosisCertificate(ctx, authApp, &publish)
+					verifiedClaims, err := verifier.VerifyDiagnosisCertificate(ctx, authApp, &publish, tc.EnforceAudience)
 					if err != nil {
-						if !strings.Contains(err.Error(), tc.Error) {
+						if tc.Error == "" {
+							t.Fatalf("unexpected error: %v", err)
+						} else if !strings.Contains(err.Error(), tc.Error) {
 							t.Fatalf("wanted error '%v', got error '%v'", tc.Error, err.Error())
 						}
 					} else if tc.Error != "" {