Implement roaring bitmaps for historical data collection (#45709)

<!-- Add the related story/sub-task/bug number, like Resolves #123, or
remove if NA -->
**Related issue:** Resolves #45715 

# Details

This PR refactors the way the charts module stores historical data to
use the [roaring bitmap](https://github.com/RoaringBitmap/roaring)
package instead of saving raw bitmaps. See [this
blurb](https://github.com/RoaringBitmap/roaring#how-does-roaring-compares-with-the-alternatives)
to learn how roaring compresses data, but TL;DR for our purposes it
represents a huge improvement especially for larger deployments where
host ID numbers may be very large. In testing, some data was reduced
96%.

The majority of the changes in this PR are straight swapping of types
from `[]byte` to `*roaring.Bitmap` in vars and function signatures, and
updating the internals of our bit math helpers to use roaring methods
instead of native AND and OR methods. I've tried to comment on all
functional changes.

Since the charts have been shipped already, so there will be data in the
wild in the prior "dense" format, the code still handles dense bitmaps
on _read_, but will always _write_ roaring bitmaps. The majority of the
data will therefore have turned over within 30 days on its own, but I
plan on a follow-up PR that will transform open rows when the cron runs
so that we should be guaranteed to turn over completely within 30 days.

# Checklist for submitter

If some of the following don't apply, delete the relevant line.

- [X] Changes file added for user-visible changes in `changes/`,
`orbit/changes/` or `ee/fleetd-chrome/changes`.
See [Changes
files](https://github.com/fleetdm/fleet/blob/main/docs/Contributing/guides/committing-changes.md#changes-files)
for more information.

- [X] Input data is properly validated, `SELECT *` is avoided, SQL
injection is prevented (using placeholders for values in statements), JS
inline code is prevented especially for url redirects, and untrusted
data interpolated into shell scripts/commands is validated against shell
metacharacters.

## Testing

- [X] Added/updated automated tests
- Tests updated to accommodate the new format, and existing unchanged
tests act as proof against regression
- [X] QA'd all new/changed functionality manually
- Using a tool that dumps the `host_scd_data` rows data into a JSON file
(with the keys being entity_id+data and the values being host IDs on
that date), compared the data from main branch and this and confirmed
they're identical
- With a host count of ~9000, some of which have IDs of over 1,000,000,
the data storage requirements were:
     * 82,558,976 bytes for dense
     * 2,867,200 for roaring (a 96% decrease)

For unreleased bug fixes in a release candidate, one of:

- [X] Confirmed that the fix is not expected to adversely impact load
test results
  - should hugely improve
- [X] Alerted the release DRI if additional load testing is needed

## Database migrations

- [X] Checked schema for all modified table for columns that will
auto-update timestamps during migration.


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## Summary by CodeRabbit

* **New Features**
* Implemented roaring bitmaps in historical data collection to optimize
bitmap handling for chart data aggregation
* Added encoding support to bitmap storage schema for flexible data
representation

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

tools/charts-backfill/README.md

@@ -1,14 +1,15 @@
 # charts-backfill
 
 Generates synthetic chart data for development and testing. Writes rows to
-`host_hourly_data_blobs` using `ON DUPLICATE KEY UPDATE`, so it is safe to
-re-run.
+`host_scd_data` using `ON DUPLICATE KEY UPDATE`, so it is safe to re-run.
+All writes use the roaring bitmap encoding (`encoding_type = 1`).
 
 ## Usage
 
 ```bash
 go run ./tools/charts-backfill --dataset uptime --days 30
 go run ./tools/charts-backfill --dataset uptime --days 7 --host-ids 1,2,3
+go run ./tools/charts-backfill --dataset cve --days 30 --use-tracked-cves
 go run ./tools/charts-backfill --dataset cve --days 30 --entity-ids CVE-2024-1,CVE-2024-2
 go run ./tools/charts-backfill --mysql-dsn "fleet:fleet@tcp(localhost:3306)/fleet"
 ```
@@ -21,13 +22,28 @@ go run ./tools/charts-backfill --mysql-dsn "fleet:fleet@tcp(localhost:3306)/flee
 | `--days` | `30` | Number of days to backfill |
 | `--start-date` | `now - days` | Start date (`YYYY-MM-DD`) |
 | `--entity-ids` | `""` | Comma-separated entity IDs (e.g. CVE IDs); `""` for non-entity datasets |
+| `--use-tracked-cves` | `false` | For `--dataset cve`, auto-discover entity IDs from the production tracked-CVE query (joins `software_cve` / `operating_system_vulnerabilities` against the curated software matchers; requires vulnerability data to be populated). Overrides `--entity-ids`. |
 | `--host-ids` | all hosts | Comma-separated host IDs to include |
 | `--mysql-dsn` | local dev | MySQL connection string |
 
 ## Datasets
 
-- **Hourly blob** (default): 24 rows/day per entity, one per hour.
-- **Daily blob** (`cve`): one row/day with `hour = -1` (whole-day sentinel).
+Backfill mode matches the live collector's sample strategy for each dataset:
 
-Density (fraction of hosts marked active) varies by dataset — see
-`densityRange` in `main.go`.
+- **Accumulate, hourly** (default; `uptime`, `policy`): 24 independent rows
+  per day per entity, each a fresh random sample. `valid_to` is set to one
+  hour past `valid_from`.
+
+- **Snapshot, state-segment** (`cve`): per-entity state-segment rows shaped
+  like real CVE data. Each entity gets an initial host set; for each
+  subsequent day, with ~5% probability the set is *churned* (~10% drop, ~10%
+  add). Each contiguous run of unchanged days collapses to a single row.
+  The final segment per entity leaves `valid_to` at the open sentinel so the
+  live collector compares against it on its next tick instead of inserting
+  over the top. Pair with `--use-tracked-cves` to mirror production CVE
+  selection.
+
+Density (fraction of hosts marked active) for the initial sample varies by
+dataset — see `densityRange` in `main.go`. Snapshot churn parameters
+(`snapshotFlipsPerDayPerEntity`, `snapshotChurnFraction`) are also defined
+there.

tools/charts-backfill/main.go

@@ -1,37 +1,99 @@
 // charts-backfill generates realistic chart data for development and testing.
-// Writes rows to host_scd_data in closed form (explicit valid_to); the live
-// collector can then extend from these rows via its normal write path.
-// Safe to re-run — uses ON DUPLICATE KEY UPDATE to merge new data.
+// Writes rows to host_scd_data. Safe to re-run — uses ON DUPLICATE KEY UPDATE
+// to merge new data.
+//
+// Datasets are backfilled in one of two modes based on their sample strategy:
+//
+//   - Accumulate (e.g. uptime): independent rows per hour, each a fresh
+//     random sample. Each row's validity is bounded to its single hour.
+//   - Snapshot (e.g. cve): per-entity state-segment rows. Most entities get
+//     a single open row spanning the entire backfill range; a small fraction
+//     "flip" state on day boundaries, producing additional closed segments.
+//     The final segment per entity has valid_to = sentinel so the live
+//     collector can compare against it on its next tick. This mirrors what
+//     real CVE data looks like (mostly stable, occasional churn).
 //
 // Usage:
 //
 //	go run ./tools/charts-backfill --dataset uptime --days 30
 //	go run ./tools/charts-backfill --dataset uptime --days 7 --host-ids 1,2,3
+//	go run ./tools/charts-backfill --dataset cve --days 30 --use-tracked-cves
 //	go run ./tools/charts-backfill --mysql-dsn "fleet:fleet@tcp(localhost:3306)/fleet"
 package main
 
 import (
+	"context"
 	"database/sql"
 	"flag"
 	"log"
+	"log/slog"
 	"math/rand/v2"
 	"time"
 
 	"github.com/fleetdm/fleet/v4/pkg/str"
 	"github.com/fleetdm/fleet/v4/server/chart"
+	"github.com/fleetdm/fleet/v4/server/chart/bootstrap"
 	_ "github.com/go-sql-driver/mysql"
+	"github.com/jmoiron/sqlx"
 )
 
-// dailyDatasets bucket at 24h granularity; all others are hourly.
-var dailyDatasets = map[string]struct{}{
+// snapshotDatasets are generated with the state-segment model: mostly stable
+// rows with occasional churn. Everything not listed here uses the accumulate
+// hourly model. Must match the live collector's sample strategy for each
+// dataset (see server/chart/datasets.go) so backfilled data is shaped like
+// what production will eventually produce.
+var snapshotDatasets = map[string]struct{}{
 	"cve": {},
 }
 
+// scdOpenSentinel mirrors the constant in server/chart/internal/mysql/data.go.
+// Used as valid_to to mark rows as currently open; the live collector closes
+// these on the next state change or extends them by leaving them alone.
+var scdOpenSentinel = time.Date(9999, 12, 31, 0, 0, 0, 0, time.UTC)
+
+// snapshotFlipsPerDayPerEntity is the per-entity probability of a state
+// change on any given day. ~5% means a 30-day window produces on average
+// ~1.5 state changes per entity — most entities stay stable, a few have a
+// handful of segments.
+const snapshotFlipsPerDayPerEntity = 0.05
+
+// snapshotChurnFraction is the fraction of an entity's current host set that
+// turns over on a flip — some hosts drop out (patched), some new hosts get
+// added (newly discovered as affected). Cardinality stays roughly stable.
+const snapshotChurnFraction = 0.10
+
+// snapshotCardinality picks a per-entity affected-host count from a long-tail
+// distribution shaped like real-world CVE data: most CVEs touch a handful of
+// hosts (specific software/version), with an occasional wide one (browser or
+// kernel). A naive uniform-density model saturates at fleet size when many
+// CVEs are unioned together — this distribution keeps the union meaningful
+// even with hundreds of tracked entities. Return value is capped at fleetSize.
+func snapshotCardinality(fleetSize int) int {
+	r := rand.Float64() //nolint:gosec // dev data generator, not crypto
+	var count int
+	switch {
+	case r < 0.70: // very narrow: specific software build
+		count = 1 + rand.IntN(5) //nolint:gosec
+	case r < 0.92: // narrow: software version
+		count = 5 + rand.IntN(20) //nolint:gosec
+	case r < 0.99: // moderate: popular software
+		count = 25 + rand.IntN(100) //nolint:gosec
+	default: // wide: browser/kernel-tier, up to ~10% of fleet
+		wideMax := max(fleetSize/10, 200)
+		count = 125 + rand.IntN(wideMax) //nolint:gosec
+	}
+	if count > fleetSize {
+		count = fleetSize
+	}
+	return count
+}
+
 func main() {
 	dataset := flag.String("dataset", "uptime", "dataset name (e.g. uptime, policy, cve)")
 	days := flag.Int("days", 30, "number of days to backfill")
 	startDate := flag.String("start-date", "", "start date (YYYY-MM-DD), defaults to now - days")
 	entityIDsStr := flag.String("entity-ids", "", "comma-separated entity IDs (default: '' for non-entity datasets)")
+	useTrackedCVEs := flag.Bool("use-tracked-cves", false, "for --dataset cve, auto-discover entity IDs from the production tracked-CVE query (overrides --entity-ids)")
 	hostIDsStr := flag.String("host-ids", "", "comma-separated host IDs (default: all from hosts table)")
 	dsn := flag.String("mysql-dsn", "fleet:fleet@tcp(localhost:3306)/fleet?parseTime=true", "MySQL connection string")
 	flag.Parse()
@@ -48,20 +110,23 @@ func main() {
 	}
 	start = time.Date(start.Year(), start.Month(), start.Day(), 0, 0, 0, 0, time.UTC)
 
-	db, err := sql.Open("mysql", *dsn)
+	rawDB, err := sql.Open("mysql", *dsn)
 	if err != nil {
 		log.Fatalf("failed to connect to mysql: %v", err)
 	}
-
-	if err := db.Ping(); err != nil {
-		db.Close()
+	if err := rawDB.Ping(); err != nil {
+		rawDB.Close()
 		log.Fatalf("failed to ping mysql: %v", err)
 	}
-	defer db.Close()
+	defer rawDB.Close()
+
+	// sqlx wraps the raw connection so we can hand it to the chart bootstrap
+	// helpers (TrackedCriticalCVEs) without opening a second pool.
+	db := sqlx.NewDb(rawDB, "mysql")
 
 	hostIDs := str.ParseUintList(*hostIDsStr)
 	if len(hostIDs) == 0 {
-		hostIDs, err = queryHostIDs(db)
+		hostIDs, err = queryHostIDs(rawDB)
 		if err != nil {
 			log.Fatalf("failed to query host IDs: %v", err) //nolint:gocritic // dev tool, OS reclaims db handle on exit
 		}
@@ -70,8 +135,25 @@ func main() {
 		}
 	}
 
-	entityIDs := str.ParseStringList(*entityIDsStr)
-	if len(entityIDs) == 0 {
+	var entityIDs []string
+	switch {
+	case *useTrackedCVEs:
+		if *dataset != "cve" {
+			log.Fatalf("--use-tracked-cves only applies to --dataset cve (got %q)", *dataset)
+		}
+		ctx := context.Background()
+		cves, err := bootstrap.TrackedCriticalCVEs(ctx, db, slog.New(slog.DiscardHandler))
+		if err != nil {
+			log.Fatalf("failed to query tracked CVEs: %v", err)
+		}
+		if len(cves) == 0 {
+			log.Fatal("tracked-CVE query returned no CVEs (vulnerability data may not be populated yet)")
+		}
+		entityIDs = cves
+		log.Printf("discovered %d tracked CVEs from the live database", len(entityIDs))
+	case *entityIDsStr != "":
+		entityIDs = str.ParseStringList(*entityIDsStr)
+	default:
 		entityIDs = []string{""}
 	}
 
@@ -79,13 +161,13 @@ func main() {
 		*dataset, *days, start.Format("2006-01-02"), len(hostIDs), len(entityIDs))
 
 	startTime := time.Now()
-	totalRows := backfill(db, *dataset, *days, start, hostIDs, entityIDs)
+	totalRows := backfill(rawDB, *dataset, *days, start, hostIDs, entityIDs)
 	log.Printf("done: %d SCD rows inserted/updated in %.1fs", totalRows, time.Since(startTime).Seconds())
 }
 
 func backfill(db *sql.DB, dataset string, days int, start time.Time, hostIDs []uint, entityIDs []string) int {
-	if _, ok := dailyDatasets[dataset]; ok {
-		return backfillDaily(db, dataset, days, start, hostIDs, entityIDs)
+	if _, ok := snapshotDatasets[dataset]; ok {
+		return backfillSnapshot(db, dataset, days, start, hostIDs, entityIDs)
 	}
 	return backfillHourly(db, dataset, days, start, hostIDs, entityIDs)
 }
@@ -107,10 +189,10 @@ func backfillHourly(db *sql.DB, dataset string, days int, start time.Time, hostI
 				blob := chart.HostIDsToBlob(activeHosts)
 
 				_, err := db.Exec(
-					`INSERT INTO host_scd_data (dataset, entity_id, host_bitmap, valid_from, valid_to)
-					 VALUES (?, ?, ?, ?, ?)
-					 ON DUPLICATE KEY UPDATE host_bitmap = VALUES(host_bitmap), valid_to = VALUES(valid_to)`,
-					dataset, entityID, blob, validFrom, validTo)
+					`INSERT INTO host_scd_data (dataset, entity_id, host_bitmap, encoding_type, valid_from, valid_to)
+					 VALUES (?, ?, ?, ?, ?, ?)
+					 ON DUPLICATE KEY UPDATE host_bitmap = VALUES(host_bitmap), encoding_type = VALUES(encoding_type), valid_to = VALUES(valid_to)`,
+					dataset, entityID, blob.Bytes, blob.Encoding, validFrom, validTo)
 				if err != nil {
 					log.Fatalf("insert hourly SCD row failed on %s hour %d: %v", validFrom, hour, err)
 				}
@@ -126,48 +208,134 @@ func backfillHourly(db *sql.DB, dataset string, days int, start time.Time, hostI
 	return totalRows
 }
 
-func backfillDaily(db *sql.DB, dataset string, days int, start time.Time, hostIDs []uint, entityIDs []string) int {
+// backfillSnapshot models per-entity state-segment data shaped like what the
+// production snapshot collector produces over time. For each entity:
+//   - Pick an initial host set with density in the dataset's typical range.
+//   - Walk day by day; on each day, with probability snapshotFlipsPerDayPerEntity,
+//     churn the set (drop ~churn% / add ~churn% of new hosts).
+//   - Each contiguous run of unchanged days is written as a single row.
+//   - The final segment per entity leaves valid_to at the sentinel (open), so
+//     the live collector compares against it on its next tick rather than
+//     opening a fresh row over the top.
+func backfillSnapshot(db *sql.DB, dataset string, days int, start time.Time, hostIDs []uint, entityIDs []string) int {
 	totalRows := 0
-	minDensity, maxDensity := densityRange(dataset)
-	n := len(hostIDs)
 
-	for day := range days {
-		date := start.AddDate(0, 0, day)
+	type segment struct {
+		validFrom time.Time
+		active    []uint
+	}
 
-		for _, entityID := range entityIDs {
-			density := minDensity + rand.Float64()*(maxDensity-minDensity) //nolint:gosec // dev data generator, not crypto
-			count := int(float64(n) * density)
-			if count == 0 {
+	for entityIdx, entityID := range entityIDs {
+		active := randomSubset(hostIDs, snapshotCardinality(len(hostIDs)))
+
+		segments := []segment{{validFrom: start, active: active}}
+		for day := 1; day < days; day++ {
+			if rand.Float64() >= snapshotFlipsPerDayPerEntity { //nolint:gosec // dev data generator, not crypto
 				continue
 			}
-			active := make([]uint, count)
-			for i, idx := range rand.Perm(n)[:count] {
-				active[i] = hostIDs[idx]
+			active = churn(active, hostIDs, snapshotChurnFraction)
+			segments = append(segments, segment{
+				validFrom: start.AddDate(0, 0, day),
+				active:    active,
+			})
+		}
+
+		for i, seg := range segments {
+			validTo := scdOpenSentinel
+			if i+1 < len(segments) {
+				validTo = segments[i+1].validFrom
 			}
-			blob := chart.HostIDsToBlob(active)
-			validFrom := date
-			validTo := date.AddDate(0, 0, 1)
+			blob := chart.HostIDsToBlob(seg.active)
 
 			_, err := db.Exec(
-				`INSERT INTO host_scd_data (dataset, entity_id, host_bitmap, valid_from, valid_to)
-				 VALUES (?, ?, ?, ?, ?)
-				 ON DUPLICATE KEY UPDATE host_bitmap = VALUES(host_bitmap), valid_to = VALUES(valid_to)`,
-				dataset, entityID, blob, validFrom, validTo)
+				`INSERT INTO host_scd_data (dataset, entity_id, host_bitmap, encoding_type, valid_from, valid_to)
+				 VALUES (?, ?, ?, ?, ?, ?)
+				 ON DUPLICATE KEY UPDATE host_bitmap = VALUES(host_bitmap), encoding_type = VALUES(encoding_type), valid_to = VALUES(valid_to)`,
+				dataset, entityID, blob.Bytes, blob.Encoding, seg.validFrom, validTo)
 			if err != nil {
-				log.Fatalf("insert daily SCD row failed on %s entity %q: %v", date, entityID, err)
+				log.Fatalf("insert snapshot SCD row failed for entity %q at %s: %v", entityID, seg.validFrom, err)
 			}
 			totalRows++
 		}
 
-		if (day+1)%5 == 0 || day == days-1 {
-			log.Printf("  day %d/%d (%s) — %d rows so far",
-				day+1, days, date.Format("2006-01-02"), totalRows)
+		if (entityIdx+1)%500 == 0 || entityIdx == len(entityIDs)-1 {
+			log.Printf("  entity %d/%d — %d rows so far",
+				entityIdx+1, len(entityIDs), totalRows)
 		}
 	}
 
 	return totalRows
 }
 
+// randomSubset returns a uniformly random `count`-sized subset of pool. If
+// count >= len(pool), returns a shuffled clone of the entire pool. The result
+// is a fresh slice that the caller can mutate.
+func randomSubset(pool []uint, count int) []uint {
+	if count <= 0 {
+		return nil
+	}
+	if count >= len(pool) {
+		out := make([]uint, len(pool))
+		copy(out, pool)
+		return out
+	}
+	out := make([]uint, count)
+	for i, idx := range rand.Perm(len(pool))[:count] { //nolint:gosec // dev data generator, not crypto
+		out[i] = pool[idx]
+	}
+	return out
+}
+
+// churn produces a new host set from `prev` by dropping a `fraction` of its
+// members and adding a `fraction` of currently-unaffected hosts from the pool.
+// Cardinality stays roughly stable; identity shifts. Models the realistic CVE
+// state-change pattern (some hosts patched, some new hosts discovered as
+// affected) without inventing wholly new bitmaps.
+func churn(prev, pool []uint, fraction float64) []uint {
+	prevSet := make(map[uint]struct{}, len(prev))
+	for _, id := range prev {
+		prevSet[id] = struct{}{}
+	}
+
+	dropCount := int(float64(len(prev)) * fraction)
+	if dropCount < 1 && len(prev) > 0 {
+		dropCount = 1
+	}
+	addCount := dropCount
+
+	// Drop: keep prev members not in the random dropout sample. If dropCount
+	// >= len(prev) we drop everything, leaving kept empty for the add step
+	// below to fill.
+	kept := make([]uint, 0, len(prev))
+	if dropCount < len(prev) {
+		dropIdx := make(map[int]struct{}, dropCount)
+		for _, idx := range rand.Perm(len(prev))[:dropCount] { //nolint:gosec // dev data generator, not crypto
+			dropIdx[idx] = struct{}{}
+		}
+		for i, id := range prev {
+			if _, drop := dropIdx[i]; drop {
+				continue
+			}
+			kept = append(kept, id)
+		}
+	}
+
+	// Add: walk a shuffled pool, picking hosts that aren't already in prev.
+	added := 0
+	for _, idx := range rand.Perm(len(pool)) { //nolint:gosec // dev data generator, not crypto
+		if added >= addCount {
+			break
+		}
+		candidate := pool[idx]
+		if _, exists := prevSet[candidate]; exists {
+			continue
+		}
+		kept = append(kept, candidate)
+		added++
+	}
+	return kept
+}
+
 // generateHourlyHosts returns a map of hour -> active host IDs for a single day.
 func generateHourlyHosts(dataset string, hostIDs []uint) map[int][]uint {
 	minDensity, maxDensity := densityRange(dataset)

tools/charts-collect/main.go

@@ -25,7 +25,6 @@
 package main
 
 import (
-	"bytes"
 	"database/sql"
 	"encoding/json"
 	"flag"
@@ -36,6 +35,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/RoaringBitmap/roaring"
 	"github.com/fleetdm/fleet/v4/pkg/fleethttp"
 	"github.com/fleetdm/fleet/v4/server/chart"
 	_ "github.com/go-sql-driver/mysql"
@@ -188,30 +188,36 @@ func collectUptime(api *apiClient, db *sql.DB) error {
 	now := time.Now().UTC()
 	bucketStart := now.Truncate(time.Hour)
 	validTo := bucketStart.Add(time.Hour)
-	newBlob := chart.HostIDsToBlob(hostIDs)
+	merged := chart.NewBitmap(hostIDs)
 
 	// OR with existing in-bucket bitmap (accumulate semantic).
-	var existing []byte
+	var existingBytes []byte
+	var existingEncoding uint8
 	err = db.QueryRow(
-		`SELECT host_bitmap FROM host_scd_data
+		`SELECT host_bitmap, encoding_type FROM host_scd_data
 		 WHERE dataset = 'uptime' AND entity_id = '' AND valid_from = ?`,
 		bucketStart,
-	).Scan(&existing)
+	).Scan(&existingBytes, &existingEncoding)
 	if err == nil {
-		newBlob = chart.BlobOR(existing, newBlob)
+		existing, decErr := chart.DecodeBitmap(chart.Blob{Bytes: existingBytes, Encoding: existingEncoding})
+		if decErr != nil {
+			return fmt.Errorf("decode existing uptime bitmap: %w", decErr)
+		}
+		merged = chart.BlobOR(merged, existing)
 	}
 
+	blob := chart.BitmapToBlob(merged)
 	_, err = db.Exec(
-		`INSERT INTO host_scd_data (dataset, entity_id, host_bitmap, valid_from, valid_to)
-		 VALUES ('uptime', '', ?, ?, ?)
-		 ON DUPLICATE KEY UPDATE host_bitmap = VALUES(host_bitmap)`,
-		newBlob, bucketStart, validTo,
+		`INSERT INTO host_scd_data (dataset, entity_id, host_bitmap, encoding_type, valid_from, valid_to)
+		 VALUES ('uptime', '', ?, ?, ?, ?)
+		 ON DUPLICATE KEY UPDATE host_bitmap = VALUES(host_bitmap), encoding_type = VALUES(encoding_type)`,
+		blob.Bytes, blob.Encoding, bucketStart, validTo,
 	)
 	if err != nil {
 		return fmt.Errorf("write uptime SCD row: %w", err)
 	}
 
-	log.Printf("  wrote uptime row: %d hosts, valid_from %s", chart.BlobPopcount(newBlob), bucketStart)
+	log.Printf("  wrote uptime row: %d hosts, valid_from %s", chart.BlobPopcount(merged), bucketStart)
 	return nil
 }
 
@@ -246,9 +252,9 @@ func collectCVE(api *apiClient, db *sql.DB) error {
 	log.Printf("  %d unique CVEs found in %.1fs", len(cveHosts), time.Since(fetchStart).Seconds())
 
 	// Build the desired entity->bitmap map for the current hourly bucket.
-	entityBitmaps := make(map[string][]byte, len(cveHosts))
+	entityBitmaps := make(map[string]*roaring.Bitmap, len(cveHosts))
 	for cve, hosts := range cveHosts {
-		entityBitmaps[cve] = chart.HostIDsToBlob(hosts)
+		entityBitmaps[cve] = chart.NewBitmap(hosts)
 	}
 
 	// Snapshot rows are keyed to 1h boundaries (not 24h) so that row transitions
@@ -267,9 +273,9 @@ func collectCVE(api *apiClient, db *sql.DB) error {
 
 // reconcileSnapshot mirrors Datastore.recordSnapshot in
 // server/chart/internal/mysql/data.go.
-func reconcileSnapshot(db *sql.DB, dataset string, entityBitmaps map[string][]byte, bucketStart time.Time) error {
+func reconcileSnapshot(db *sql.DB, dataset string, entityBitmaps map[string]*roaring.Bitmap, bucketStart time.Time) error {
 	rows, err := db.Query(
-		`SELECT entity_id, host_bitmap, valid_from
+		`SELECT entity_id, host_bitmap, encoding_type, valid_from
 		 FROM host_scd_data
 		 WHERE dataset = ? AND valid_to = ?`,
 		dataset, scdOpenSentinel)
@@ -277,42 +283,45 @@ func reconcileSnapshot(db *sql.DB, dataset string, entityBitmaps map[string][]by
 		return fmt.Errorf("fetch open SCD rows: %w", err)
 	}
 	defer rows.Close()
-	type openRow struct {
-		bitmap    []byte
+	type openEntry struct {
+		bitmap    *roaring.Bitmap
 		validFrom time.Time
 	}
-	openByEntity := make(map[string]openRow)
+	openByEntity := make(map[string]openEntry)
 	for rows.Next() {
 		var entityID string
-		var bitmap []byte
+		var bitmapBytes []byte
+		var encoding uint8
 		var validFrom time.Time
-		if err := rows.Scan(&entityID, &bitmap, &validFrom); err != nil {
+		if err := rows.Scan(&entityID, &bitmapBytes, &encoding, &validFrom); err != nil {
 			return fmt.Errorf("scan open SCD row: %w", err)
 		}
-		openByEntity[entityID] = openRow{bitmap: bitmap, validFrom: validFrom}
+		rb, err := chart.DecodeBitmap(chart.Blob{Bytes: bitmapBytes, Encoding: encoding})
+		if err != nil {
+			return fmt.Errorf("decode open bitmap for %q: %w", entityID, err)
+		}
+		openByEntity[entityID] = openEntry{bitmap: rb, validFrom: validFrom}
 	}
 	if err := rows.Err(); err != nil {
 		return fmt.Errorf("iterate open SCD rows: %w", err)
 	}
 
 	var toClose []string
-	var toUpsert []struct {
+	type upsertRow struct {
 		entityID string
-		bitmap   []byte
+		blob     chart.Blob
 	}
+	var toUpsert []upsertRow
 
 	for entityID, bitmap := range entityBitmaps {
 		existing, hasOpen := openByEntity[entityID]
-		if hasOpen && bytes.Equal(existing.bitmap, bitmap) {
+		if hasOpen && existing.bitmap.Equals(bitmap) {
 			continue
 		}
 		if hasOpen && existing.validFrom.Before(bucketStart) {
 			toClose = append(toClose, entityID)
 		}
-		toUpsert = append(toUpsert, struct {
-			entityID string
-			bitmap   []byte
-		}{entityID, bitmap})
+		toUpsert = append(toUpsert, upsertRow{entityID: entityID, blob: chart.BitmapToBlob(bitmap)})
 	}
 
 	for entityID := range openByEntity {
@@ -343,15 +352,15 @@ func reconcileSnapshot(db *sql.DB, dataset string, entityBitmaps map[string][]by
 		batch := toUpsert[i:end]
 
 		placeholders := make([]string, len(batch))
-		args := make([]any, 0, len(batch)*4)
+		args := make([]any, 0, len(batch)*5)
 		for j, r := range batch {
-			placeholders[j] = "(?, ?, ?, ?)"
-			args = append(args, dataset, r.entityID, r.bitmap, bucketStart)
+			placeholders[j] = "(?, ?, ?, ?, ?)"
+			args = append(args, dataset, r.entityID, r.blob.Bytes, r.blob.Encoding, bucketStart)
 		}
-		// Concatenating hardcoded "(?,?,?,?)" placeholder strings, not user input.
-		stmt := `INSERT INTO host_scd_data (dataset, entity_id, host_bitmap, valid_from) VALUES ` + //nolint:gosec // G202
+		// Concatenating hardcoded "(?,?,?,?,?)" placeholder strings, not user input.
+		stmt := `INSERT INTO host_scd_data (dataset, entity_id, host_bitmap, encoding_type, valid_from) VALUES ` + //nolint:gosec // G202
 			strings.Join(placeholders, ", ") +
-			` ON DUPLICATE KEY UPDATE host_bitmap = VALUES(host_bitmap)`
+			` ON DUPLICATE KEY UPDATE host_bitmap = VALUES(host_bitmap), encoding_type = VALUES(encoding_type)`
 		if _, err := db.Exec(stmt, args...); err != nil {
 			return fmt.Errorf("upsert rows: %w", err)
 		}