From 9e62fba0ad5699f8fec2ca60b8a657b30dcd8265 Mon Sep 17 00:00:00 2001 From: Brad Lugo Date: Wed, 26 Aug 2026 01:25:07 -0700 Subject: [PATCH 1/4] postgres: add alias link test and benchmark Exercise UpdateVulnerabilities with enough alias-carrying vulnerabilities to cross batch-flush boundaries and verify the link tables directly. The benchmark reports peak live heap alongside the usual metrics, since allocation lifetime is invisible to B/op. Signed-off-by: Brad Lugo Signed-off-by: Hank Donnay Change-Id: Id51fc15a093d90e78733e31182a631e86a6a6964 --- .../updatevulnerabilities_benchmark_test.go | 73 +++++++++++++++++ .../postgres/updatevulnerabilities_test.go | 78 +++++++++++++++++++ 2 files changed, 151 insertions(+) create mode 100644 datastore/postgres/updatevulnerabilities_benchmark_test.go diff --git a/datastore/postgres/updatevulnerabilities_benchmark_test.go b/datastore/postgres/updatevulnerabilities_benchmark_test.go new file mode 100644 index 000000000..8ac028fc5 --- /dev/null +++ b/datastore/postgres/updatevulnerabilities_benchmark_test.go @@ -0,0 +1,73 @@ +package postgres + +import ( + "runtime" + "runtime/metrics" + "strconv" + "sync" + "testing" + "time" + + "github.com/google/uuid" + + "github.com/quay/claircore/libvuln/driver" + "github.com/quay/claircore/test" + "github.com/quay/claircore/test/integration" + pgtest "github.com/quay/claircore/test/postgres" +) + +func sampleValue(s []metrics.Sample) uint64 { + metrics.Read(s) + return s[0].Value.Uint64() +} + +func BenchmarkUpdateVulnerabilities(b *testing.B) { + integration.NeedDB(b) + // Consider using `-benchtime 1x` when running the 50000, 75000, and 100000. + for _, sz := range []int{100, 500, 1200, 50000, 75000, 100000} { + b.Run(strconv.Itoa(sz)+"Vulnerabilities", func(b *testing.B) { + ctx := test.Logging(b) + pool := pgtest.TestMatcherDB(ctx, b) + store := NewMatcherStore(pool) + vulns := genAliasVulns(b.Name(), sz) + + // Sample live heap during the run: B/op reports cumulative + // allocation and misses how long allocations stay reachable, + // which is what the chunked alias flush is meant to bound. + runtime.GC() + done := make(chan struct{}) + var wg sync.WaitGroup + wg.Go(func() { + // Record the (lagging) size of the live heap with this metric. + sample := []metrics.Sample{{Name: "/gc/heap/live:bytes"}} + base := sampleValue(sample) + var peak uint64 + tick := time.NewTicker(25 * time.Millisecond) + defer tick.Stop() + Tick: + for { + select { + case <-done: + break Tick + case <-tick.C: + peak = max(peak, sampleValue(sample)) + } + } + runtime.GC() + peak = max(peak, sampleValue(sample)) + b.ReportMetric(float64(peak-base), "heapGrowth-B") + }) + + b.ReportAllocs() + b.ResetTimer() + for b.Loop() { + if _, err := store.UpdateVulnerabilities(ctx, b.Name(), driver.Fingerprint(uuid.New().String()), vulns); err != nil { + b.Fatalf("UpdateVulnerabilities: %v", err) + } + } + b.StopTimer() + close(done) + wg.Wait() + }) + } +} diff --git a/datastore/postgres/updatevulnerabilities_test.go b/datastore/postgres/updatevulnerabilities_test.go index 505e21614..c066401bb 100644 --- a/datastore/postgres/updatevulnerabilities_test.go +++ b/datastore/postgres/updatevulnerabilities_test.go @@ -1,6 +1,7 @@ package postgres import ( + "fmt" "sync" "testing" "unique" @@ -399,3 +400,80 @@ func TestUpdateVulnerabilitiesIterSinglePass(t *testing.T) { } t.Logf("vuln rows: %d, alias rows: %d", vulnCount, aliasCount) } + +// SharedAliasName is attached to every vulnerability generated by +// genAliasVulns, so it recurs in every flush chunk. +const sharedAliasName = "GHSA-shared-0000-0000" + +func genAliasVulns(updater string, n int) []*claircore.Vulnerability { + shared := claircore.Alias{Space: unique.Make("GHSA"), Name: sharedAliasName} + vulns := make([]*claircore.Vulnerability, n) + for i := range vulns { + name := fmt.Sprintf("CVE-2024-%04d", i) + vulns[i] = &claircore.Vulnerability{ + Updater: updater, + Name: name, + Package: &claircore.Package{Name: "test-pkg"}, + Self: claircore.Alias{Space: unique.Make("CVE"), Name: name}, + Aliases: []claircore.Alias{ + shared, + {Space: unique.Make("GHSA"), Name: "GHSA-" + name}, + }, + } + } + return vulns +} + +func TestUpdateVulnerabilitiesChunked(t *testing.T) { + integration.NeedDB(t) + ctx := test.Logging(t) + + pool := pgtest.TestMatcherDB(ctx, t) + store := NewMatcherStore(pool) + + // The insert batch flushes every 500 vulnerabilities (two queued queries + // per vulnerability, flushed at 1000), so 1200 crosses two chunk + // boundaries and leaves a remainder for the final flush. + const vulnCt = 1200 + vulns := genAliasVulns(t.Name(), vulnCt) + + if _, err := store.UpdateVulnerabilities(ctx, t.Name(), driver.Fingerprint(uuid.New().String()), vulns); err != nil { + t.Fatalf("UpdateVulnerabilities: %v", err) + } + + checks := []struct { + desc string + query string + want int + }{ + {"vuln rows", `SELECT count(*) FROM vuln WHERE updater = $1`, vulnCt}, + {"self links", `SELECT count(*) FROM vulnerability_self s JOIN vuln v ON s.vulnerability = v.id WHERE v.updater = $1`, vulnCt}, + // One shared alias plus one unique alias per vulnerability. + {"alias links", `SELECT count(*) FROM vulnerability_alias a JOIN vuln v ON a.vulnerability = v.id WHERE v.updater = $1`, 2 * vulnCt}, + } + for _, c := range checks { + var got int + if err := pool.QueryRow(ctx, c.query, t.Name()).Scan(&got); err != nil { + t.Fatalf("counting %s: %v", c.desc, err) + } + if got != c.want { + t.Errorf("%s: got %d, want %d", c.desc, got, c.want) + } + } + + // The shared alias row is created during the first chunk; vulnerabilities + // in later chunks must still link to it. + var sharedCt int + err := pool.QueryRow(ctx, ` + SELECT count(*) + FROM vulnerability_alias va + JOIN vuln v ON va.vulnerability = v.id + JOIN alias a ON va.alias = a.id + WHERE v.updater = $1 AND a.name = $2`, t.Name(), sharedAliasName).Scan(&sharedCt) + if err != nil { + t.Fatalf("counting shared alias links: %v", err) + } + if sharedCt != vulnCt { + t.Errorf("shared alias links: got %d, want %d", sharedCt, vulnCt) + } +} From fbb932e459e4eec6ed98695e1a5c117f29d2c96b Mon Sep 17 00:00:00 2001 From: Brad Lugo Date: Wed, 26 Aug 2026 08:27:08 -0700 Subject: [PATCH 2/4] postgres: pin update transaction isolation The chunked link statements are INSERT..SELECTs that must see alias rows committed by concurrent updaters after the transaction began, so pin read committed instead of inheriting default_transaction_isolation. Signed-off-by: Brad Lugo Signed-off-by: Hank Donnay Change-Id: I462ca8ea288011295a2b3cb92ce8ae636a6a6964 --- datastore/postgres/updatevulnerabilities.go | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/datastore/postgres/updatevulnerabilities.go b/datastore/postgres/updatevulnerabilities.go index 3009e25fd..93b10119f 100644 --- a/datastore/postgres/updatevulnerabilities.go +++ b/datastore/postgres/updatevulnerabilities.go @@ -196,7 +196,11 @@ ON CONFLICT DO NOTHING;` start := time.Now() - tx, err := s.pool.Begin(ctx) + // The isolation level must be pinned to "read committed" (rather than + // inheriting default_transaction_isolation) because some INSERT ... SELECT + // statements need to see alias rows committed outside this transaction + // after it began. + tx, err := s.pool.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.ReadCommitted}) if err != nil { return uuid.Nil, fmt.Errorf("unable to start transaction: %w", err) } From 1d135644ffcf6061a3d4337cd0be7d7302836977 Mon Sep 17 00:00:00 2001 From: Hank Donnay Date: Fri, 7 Aug 2026 10:57:17 -0500 Subject: [PATCH 3/4] postgres/types: add support for encoding `unique.Handle[string]` Signed-off-by: Hank Donnay Change-Id: I85a31f5796751b32e698b7b0108461146a6a6964 --- datastore/postgres/types/types.go | 1 + datastore/postgres/types/unique_string.go | 45 +++++++++++++++++++++++ 2 files changed, 46 insertions(+) create mode 100644 datastore/postgres/types/unique_string.go diff --git a/datastore/postgres/types/types.go b/datastore/postgres/types/types.go index 3eed3ed57..7b4bd28f1 100644 --- a/datastore/postgres/types/types.go +++ b/datastore/postgres/types/types.go @@ -13,6 +13,7 @@ func ConnectRegisterTypes(ctx context.Context, c *pgx.Conn) error { for _, f := range []func(context.Context, *pgx.Conn) error{ registerVersionRange, registerPackageKind, + registerUniqueString, } { if err := f(ctx, c); err != nil { return err diff --git a/datastore/postgres/types/unique_string.go b/datastore/postgres/types/unique_string.go new file mode 100644 index 000000000..5622f3fb9 --- /dev/null +++ b/datastore/postgres/types/unique_string.go @@ -0,0 +1,45 @@ +package types + +import ( + "context" + "unique" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgtype" +) + +func registerUniqueString(ctx context.Context, c *pgx.Conn) error { + tm := c.TypeMap() + tm.TryWrapEncodePlanFuncs = append([]pgtype.TryWrapEncodePlanFunc{ + uniqueStringWrapEncodePlan, + }, tm.TryWrapEncodePlanFuncs...) + return nil +} + +func uniqueStringWrapEncodePlan(value any) (pgtype.WrappedEncodePlanNextSetter, any, bool) { + switch v := value.(type) { + case []unique.Handle[string]: + return &wrapUniqueStringSliceEncodePlan{}, pgtype.FlatArray[unique.Handle[string]](v), true + case unique.Handle[string]: + return &wrapUniqueStringEncodePlan{}, "", true + } + return nil, nil, false +} + +type wrapUniqueStringSliceEncodePlan struct { + encodeWrapper +} + +// Encode implements [pgtype.WrappedEncodePlanNextSetter]. +func (p *wrapUniqueStringSliceEncodePlan) Encode(value any, buf []byte) (newBuf []byte, err error) { + return p.next.Encode(pgtype.FlatArray[unique.Handle[string]](value.([]unique.Handle[string])), buf) +} + +type wrapUniqueStringEncodePlan struct { + encodeWrapper +} + +// Encode implements [pgtype.WrappedEncodePlanNextSetter]. +func (p *wrapUniqueStringEncodePlan) Encode(value any, buf []byte) (newBuf []byte, err error) { + return p.next.Encode(value.(unique.Handle[string]).Value(), buf) +} From 5b55a27dee9794632cd1f8232321421983278758 Mon Sep 17 00:00:00 2001 From: Hank Donnay Date: Fri, 28 Aug 2026 09:00:06 -0500 Subject: [PATCH 4/4] postgres: do callback hell instead of giant slices Signed-off-by: Hank Donnay Change-Id: I7b0aef57f26107254a546add7442b9556a6a6964 --- datastore/postgres/generate.go | 2 + ...lities_associate_update_operation_vuln.sql | 5 + .../updatevulnerabilities_insert_alias.sql | 10 + ...vulnerabilities_insert_alias_namespace.sql | 5 + .../updatevulnerabilities_insert_vuln.sql | 67 ++++ ...erabilities_insert_vulnerability_alias.sql | 5 + ...nerabilities_insert_vulnerability_self.sql | 5 + .../updatevulnerabilities_select_alias.sql | 8 + ...atevulnerabilities_select_vuln_by_hash.sql | 7 + datastore/postgres/updatevulnerabilities.go | 309 ++++++++---------- 10 files changed, 250 insertions(+), 173 deletions(-) create mode 100644 datastore/postgres/query/updatevulnerabilities_associate_update_operation_vuln.sql create mode 100644 datastore/postgres/query/updatevulnerabilities_insert_alias.sql create mode 100644 datastore/postgres/query/updatevulnerabilities_insert_alias_namespace.sql create mode 100644 datastore/postgres/query/updatevulnerabilities_insert_vuln.sql create mode 100644 datastore/postgres/query/updatevulnerabilities_insert_vulnerability_alias.sql create mode 100644 datastore/postgres/query/updatevulnerabilities_insert_vulnerability_self.sql create mode 100644 datastore/postgres/query/updatevulnerabilities_select_alias.sql create mode 100644 datastore/postgres/query/updatevulnerabilities_select_vuln_by_hash.sql diff --git a/datastore/postgres/generate.go b/datastore/postgres/generate.go index 17aa3b398..bb3d612a5 100644 --- a/datastore/postgres/generate.go +++ b/datastore/postgres/generate.go @@ -6,3 +6,5 @@ package postgres //go:generate -command mktestdata go run github.com/quay/claircore/test/bisect -dump-index "testdata/{{.}}.index.json" -dump-report "testdata/{{.}}.report.json" //go:generate mktestdata docker.io/library/amazonlinux:1 docker.io/library/debian:10 docker.io/library/debian:9 docker.io/library/debian:8 docker.io/mitmproxy/mitmproxy:4.0.1 docker.io/library/ubuntu:16.04 docker.io/library/ubuntu:18.04 docker.io/library/ubuntu:19.10 docker.io/library/ubuntu:20.04 registry.access.redhat.com/ubi8/ubi + +//go:generate find query -name *.sql -exec go run github.com/wasilibs/go-sql-formatter/v15/cmd/sql-formatter@latest --language postgresql --fix {} ; diff --git a/datastore/postgres/query/updatevulnerabilities_associate_update_operation_vuln.sql b/datastore/postgres/query/updatevulnerabilities_associate_update_operation_vuln.sql new file mode 100644 index 000000000..9052cf1b9 --- /dev/null +++ b/datastore/postgres/query/updatevulnerabilities_associate_update_operation_vuln.sql @@ -0,0 +1,5 @@ +INSERT INTO + uo_vuln (uo, vuln) +VALUES + ($1, $2) +ON CONFLICT DO NOTHING; diff --git a/datastore/postgres/query/updatevulnerabilities_insert_alias.sql b/datastore/postgres/query/updatevulnerabilities_insert_alias.sql new file mode 100644 index 000000000..46329b46e --- /dev/null +++ b/datastore/postgres/query/updatevulnerabilities_insert_alias.sql @@ -0,0 +1,10 @@ +INSERT INTO + alias (namespace, name) +SELECT + ns.id, + $2 +FROM + alias_namespace AS ns +WHERE + ns.namespace = $1 +ON CONFLICT DO NOTHING; diff --git a/datastore/postgres/query/updatevulnerabilities_insert_alias_namespace.sql b/datastore/postgres/query/updatevulnerabilities_insert_alias_namespace.sql new file mode 100644 index 000000000..02b0d99d5 --- /dev/null +++ b/datastore/postgres/query/updatevulnerabilities_insert_alias_namespace.sql @@ -0,0 +1,5 @@ +INSERT INTO + alias_namespace (namespace) +VALUES + ($1) +ON CONFLICT DO NOTHING; diff --git a/datastore/postgres/query/updatevulnerabilities_insert_vuln.sql b/datastore/postgres/query/updatevulnerabilities_insert_vuln.sql new file mode 100644 index 000000000..f8b455c87 --- /dev/null +++ b/datastore/postgres/query/updatevulnerabilities_insert_vuln.sql @@ -0,0 +1,67 @@ +INSERT INTO + vuln ( + hash_kind, + hash, + name, + updater, + description, + issued, + links, + severity, + normalized_severity, + package_name, + package_version, + package_module, + package_arch, + package_kind, + dist_id, + dist_name, + dist_version, + dist_version_code_name, + dist_version_id, + dist_arch, + dist_cpe, + dist_pretty_name, + repo_name, + repo_key, + repo_uri, + fixed_in_version, + arch_operation, + version_kind, + vulnerable_range, + not_vulnerable + ) +VALUES + ( + $1, + $2, + $3, + $4, + $5, + $6, + $7, + $8, + $9, + $10, + $11, + $12, + $13, + $14, + $15, + $16, + $17, + $18, + $19, + $20, + $21, + $22, + $23, + $24, + $25, + $26, + $27, + $28, + COALESCE($29, VersionRange ('{}', '{}', '()')), + $30 + ) +ON CONFLICT (hash_kind, hash) DO NOTHING; diff --git a/datastore/postgres/query/updatevulnerabilities_insert_vulnerability_alias.sql b/datastore/postgres/query/updatevulnerabilities_insert_vulnerability_alias.sql new file mode 100644 index 000000000..62e6e6032 --- /dev/null +++ b/datastore/postgres/query/updatevulnerabilities_insert_vulnerability_alias.sql @@ -0,0 +1,5 @@ +INSERT INTO + vulnerability_alias (vulnerability, alias) +VALUES + ($1, $2) +ON CONFLICT DO NOTHING; diff --git a/datastore/postgres/query/updatevulnerabilities_insert_vulnerability_self.sql b/datastore/postgres/query/updatevulnerabilities_insert_vulnerability_self.sql new file mode 100644 index 000000000..63b0b8fd7 --- /dev/null +++ b/datastore/postgres/query/updatevulnerabilities_insert_vulnerability_self.sql @@ -0,0 +1,5 @@ +INSERT INTO + vulnerability_self (vulnerability, self) +VALUES + ($1, $2) +ON CONFLICT DO NOTHING; diff --git a/datastore/postgres/query/updatevulnerabilities_select_alias.sql b/datastore/postgres/query/updatevulnerabilities_select_alias.sql new file mode 100644 index 000000000..9357be6b3 --- /dev/null +++ b/datastore/postgres/query/updatevulnerabilities_select_alias.sql @@ -0,0 +1,8 @@ +SELECT + a.id +FROM + alias AS a + JOIN alias_namespace AS ns ON a.namespace = ns.id +WHERE + ns.namespace = $1 + AND a.name = $2; diff --git a/datastore/postgres/query/updatevulnerabilities_select_vuln_by_hash.sql b/datastore/postgres/query/updatevulnerabilities_select_vuln_by_hash.sql new file mode 100644 index 000000000..ee049c5fd --- /dev/null +++ b/datastore/postgres/query/updatevulnerabilities_select_vuln_by_hash.sql @@ -0,0 +1,7 @@ +SELECT + id +FROM + vuln +WHERE + hash_kind = $1 + AND hash = $2; diff --git a/datastore/postgres/updatevulnerabilities.go b/datastore/postgres/updatevulnerabilities.go index 93b10119f..e8939953f 100644 --- a/datastore/postgres/updatevulnerabilities.go +++ b/datastore/postgres/updatevulnerabilities.go @@ -2,6 +2,8 @@ package postgres import ( "context" + _ "embed" // for queries + "errors" "fmt" "log/slog" "strconv" @@ -95,6 +97,27 @@ func (s *MatcherStore) DeltaUpdateVulnerabilities(ctx context.Context, updater s return s.updateVulnerabilities(ctx, updater, fingerprint, iterVulns, delVulns) } +var ( + //go:embed query/updatevulnerabilities_associate_update_operation_vuln.sql + updateVulnerabilitiesAssociateUpdateOperationVuln string + //go:embed query/updatevulnerabilities_select_vuln_by_hash.sql + updateVulnerabilitiesSelectVulnByHash string + //go:embed query/updatevulnerabilities_insert_alias_namespace.sql + updateVulnerabilitiesInsertAliasNamespace string + //go:embed query/updatevulnerabilities_insert_alias.sql + updateVulnerabilitiesInsertAlias string + //go:embed query/updatevulnerabilities_select_alias.sql + updateVulnerabilitiesSelectAlias string + //go:embed query/updatevulnerabilities_insert_vulnerability_alias.sql + updateVulnerabilitiesInsertVulnerabilityAlias string + //go:embed query/updatevulnerabilities_insert_vulnerability_self.sql + updateVulnerabilitiesInsertVulnerabilitySelf string + // Insert attempts to create a new vulnerability. It fails silently. + // + //go:embed query/updatevulnerabilities_insert_vuln.sql + updateVulnerabilitiesInsertVuln string +) + func (s *MatcherStore) updateVulnerabilities(ctx context.Context, updater string, fingerprint driver.Fingerprint, vulnIter datastore.VulnerabilityIter, delIter datastore.Iter[string]) (uuid.UUID, error) { const ( // Create makes a new update operation and returns the reference and ID. @@ -120,75 +143,7 @@ func (s *MatcherStore) updateVulnerabilities(ctx context.Context, updater string )` // assocExisting associates existing vulnerabilities with new update operations assocExisting = `INSERT INTO uo_vuln (uo, vuln) VALUES ($1, $2) ON CONFLICT DO NOTHING;` - // Insert attempts to create a new vulnerability. It fails silently. - insert = ` - INSERT INTO vuln ( - hash_kind, hash, - name, updater, description, issued, links, severity, normalized_severity, - package_name, package_version, package_module, package_arch, package_kind, - dist_id, dist_name, dist_version, dist_version_code_name, dist_version_id, dist_arch, dist_cpe, dist_pretty_name, - repo_name, repo_key, repo_uri, - fixed_in_version, arch_operation, version_kind, vulnerable_range, - not_vulnerable - ) VALUES ( - $1, $2, - $3, $4, $5, $6, $7, $8, $9, - $10, $11, $12, $13, $14, - $15, $16, $17, $18, $19, $20, $21, $22, - $23, $24, $25, - $26, $27, $28, COALESCE($29, VersionRange('{}', '{}', '()')), - $30 - ) - ON CONFLICT (hash_kind, hash) DO NOTHING;` - // Assoc associates an update operation and a vulnerability. It fails - // silently. - assoc = ` - INSERT INTO uo_vuln (uo, vuln) VALUES ( - $3, - (SELECT id FROM vuln WHERE hash_kind = $1 AND hash = $2)) - ON CONFLICT DO NOTHING;` - refreshView = `REFRESH MATERIALIZED VIEW CONCURRENTLY latest_update_operations;` - // bulkLinkAliases links all vulnerability→alias rows in one statement by - // joining the flattened (hash_kind, hash, alias_space, alias_name) arrays - // against the already-populated vuln and alias tables. - bulkLinkAliases = ` - INSERT INTO vulnerability_alias (vulnerability, alias) - SELECT v.id, a.id - FROM - unnest($1::TEXT[], $2::BYTEA[], $3::TEXT[], $4::TEXT[]) - AS input(hash_kind, hash, alias_space, alias_name) - JOIN - vuln v ON v.hash_kind = input.hash_kind AND v.hash = input.hash - JOIN - alias_namespace ns ON ns.namespace = input.alias_space - JOIN - alias a ON a.name = input.alias_name AND a.namespace = ns.id - ON CONFLICT DO NOTHING` - // bulkLinkSelf links all vulnerability→self rows in one statement. - bulkLinkSelf = ` - INSERT INTO vulnerability_self (vulnerability, self) - SELECT v.id, a.id - FROM - unnest($1::TEXT[], $2::BYTEA[], $3::TEXT[], $4::TEXT[]) - AS input(hash_kind, hash, self_space, self_name) - JOIN - vuln v ON v.hash_kind = input.hash_kind AND v.hash = input.hash - JOIN - alias_namespace ns ON ns.namespace = input.self_space - JOIN - alias a ON a.name = input.self_name AND a.namespace = ns.id - ON CONFLICT DO NOTHING` - // insertAliasNamespaces creates all needed namespace rows outside any - // transaction so concurrent updaters do not deadlock. - insertAliasNamespaces = `INSERT INTO alias_namespace (namespace) VALUES (unnest($1::TEXT[])) ON CONFLICT DO NOTHING;` - // insertAliases creates all needed alias rows outside any transaction. - insertAliases = `INSERT INTO alias (namespace, name) - SELECT ns.id, input.name - FROM - (SELECT unnest($1::TEXT[]) AS space, unnest($2::TEXT[]) AS name) AS input - JOIN - alias_namespace AS ns ON input.space = ns.namespace -ON CONFLICT DO NOTHING;` + refreshView = `REFRESH MATERIALIZED VIEW CONCURRENTLY latest_update_operations;` ) var uoID uint64 @@ -252,20 +207,18 @@ ON CONFLICT DO NOTHING;` } if len(oldVulns) > 0 { - vulnIter(func(v *claircore.Vulnerability, _ error) bool { + for v := range vulnIter { // If we have an existing vuln in the new batch // delete it from the oldVulns map so it doesn't // get associated with the new update_operation. delete(oldVulns, v.Name) - return true - }) - delIter(func(delName string, _ error) bool { + } + for delName := range delIter { // If we have an existing vuln that has been signaled // as deleted by the updater then delete it so it doesn't // get associated with the new update_operation. delete(oldVulns, delName) - return true - }) + } } start = time.Now() // Associate already existing vulnerabilities with new update_operation. @@ -279,42 +232,116 @@ ON CONFLICT DO NOTHING;` } updateVulnerabilitiesCounter.WithLabelValues("assocExisting", strconv.FormatBool(delta)).Add(float64(len(oldVulns))) updateVulnerabilitiesDuration.WithLabelValues("assocExisting", strconv.FormatBool(delta)).Observe(time.Since(start).Seconds()) - } // batch insert vulnerabilities + const batchLim = 1000 skipCt := 0 vulnCt := 0 start = time.Now() - var batch pgx.Batch + // This is an annoying way to go about this, but c'est la vie. + // + // These batches are chains of statements smeared across two database + // connections that are all connected via callbacks. + conn, err := s.pool.Acquire(ctx) + if err != nil { + return uuid.Nil, fmt.Errorf("unable to acquire alias connection: %w", err) + } + defer conn.Release() + // These are the batches used in the callback chain. The results of one are + // used to enqueue queries into the next batch. + // + // They MUST be sent in this order, and [aliasBatch] MUST be sent outside + // the transaction. + var insertBatch, aliasBatch, assocBatch pgx.Batch + // Some guesses at initial sizing. These should always level off, but + // avoiding allocations and copies is always welcome. + insertBatch.QueuedQueries = make([]*pgx.QueuedQuery, 0, batchLim+1) + aliasBatch.QueuedQueries = make([]*pgx.QueuedQuery, 0, batchLim*4) + assocBatch.QueuedQueries = make([]*pgx.QueuedQuery, 0, batchLim*5) + // Flush sends the batches in the correct order, then resets the batches' + // query slices. flush := func() (err error) { - err = tx.SendBatch(ctx, &batch).Close() - clear(batch.QueuedQueries) - batch.QueuedQueries = batch.QueuedQueries[:0] + err = errors.Join( + tx.SendBatch(ctx, &insertBatch).Close(), + conn.SendBatch(ctx, &aliasBatch).Close(), + tx.SendBatch(ctx, &assocBatch).Close(), + ) + for _, b := range []*pgx.Batch{&insertBatch, &aliasBatch, &assocBatch} { + clear(b.QueuedQueries) + b.QueuedQueries = b.QueuedQueries[:0] + } return err } - // Flattened parallel arrays for the bulk alias-linking statements run after - // all vuln inserts are done. Each entry in va* corresponds to one - // (vuln, alias) pair; each entry in vs* to one (vuln, self) pair. - var ( - vaHashKinds, vsHashKinds []string - vaHashes, vsHashes [][]byte - vaSpaces, vsSpaces []string - vaNames, vsNames []string - ) + // SeenSpace tracks alias namespaces, to avoid sending a lot of redundant + // namespace creation statements. seenSpace := make(map[unique.Handle[string]]struct{}) - seenAlias := make(map[claircore.Alias]struct{}) + // This whole function is a giant callback hell. Don't do this. I was backed + // into a corner. This function makes my son cry and actively saps joy from + // the world. + vulnIDCallback := func(vuln *claircore.Vulnerability) func(pgx.Row) error { + // VulnID is where the id for the passed-in vulnerability will be + // stored. + var vulnID uint64 + // DoAlias is a closure that enqueues the Alias insertion statements. + doAlias := func(a claircore.Alias, assoc string) { + if !a.Valid() { + return + } + if _, ok := seenSpace[a.Space]; !ok { + seenSpace[a.Space] = struct{}{} + aliasBatch.Queue(updateVulnerabilitiesInsertAliasNamespace, a.Space) + } + // It might be possible to collapse these two statements, at the + // cost of making it more complicated: INSERT ... RETURNING only + // works if an insertion happened. + aliasBatch.Queue(updateVulnerabilitiesInsertAlias, a.Space, a.Name) + aliasBatch. + Queue(updateVulnerabilitiesSelectAlias, a.Space, a.Name). + QueryRow(func(row pgx.Row) error { + // This closure enqueues the statement to associate the + // alias and the vulnerability via the correct pivot table. + var aliasID uint64 + if err := row.Scan(&aliasID); err != nil { + return err + } + if vulnID != 0 { + assocBatch.Queue(assoc, vulnID, aliasID) + } + return nil + }) + } + for _, a := range vuln.Aliases { + doAlias(a, updateVulnerabilitiesInsertVulnerabilityAlias) + } + doAlias(vuln.Self, updateVulnerabilitiesInsertVulnerabilitySelf) + // All the above should make it so that the [*claircore.Vulnerability] + // isn't pinned in memory until the batch is processed. Only the string + // backing storage and the [unique.Handle] backing storage should be + // unreclaimable while this batch is in flight. + + // As long as [insertBatch] is submitted first, [vulnID] is populated in + // this callback and the [aliasBatch] callbacks have the value to use to + // populate the [assocBatch]. + return func(row pgx.Row) error { + if err := row.Scan(&vulnID); err != nil { + return err + } + assocBatch.Queue(updateVulnerabilitiesAssociateUpdateOperationVuln, uoID, vulnID) + return nil + } + } - vulnIter(func(vuln *claircore.Vulnerability, iterErr error) bool { + for vuln, iterErr := range vulnIter { if iterErr != nil { err = iterErr - return false + break } vulnCt++ if skipVulnerability(vuln) { skipCt++ - return true + continue } pkg := vuln.Package @@ -327,52 +354,26 @@ ON CONFLICT DO NOTHING;` repo = &zeroRepo } hashKind, hash := md5Vuln(vuln) - vKind, _, _ := rangefmt(vuln.Range) - batch.Queue( - insert, + insertBatch.Queue( + updateVulnerabilitiesInsertVuln, hashKind, hash, vuln.Name, vuln.Updater, vuln.Description, vuln.Issued, vuln.Links, vuln.Severity, vuln.NormalizedSeverity, pkg.Name, pkg.Version, pkg.Module, pkg.Arch, pkg.Kind, dist.DID, dist.Name, dist.Version, dist.VersionCodeName, dist.VersionID, dist.Arch, dist.CPE, dist.PrettyName, repo.Name, repo.Key, repo.URI, - vuln.FixedInVersion, vuln.ArchOperation, vKind, vuln.Range, + vuln.FixedInVersion, vuln.ArchOperation, rangekind(vuln.Range), vuln.Range, vuln.Invert, ) - batch.Queue(assoc, hashKind, hash, uoID) + insertBatch.Queue(updateVulnerabilitiesSelectVulnByHash, hashKind, hash).QueryRow(vulnIDCallback(vuln)) - // Accumulate alias links for the bulk statements below. The hash is - // repeated once per alias so the unnest join can match each row to its - // vuln. - for _, a := range vuln.Aliases { - if !a.Valid() { - continue + if ct := insertBatch.Len(); ct >= batchLim { + if err = flush(); err != nil { + err = fmt.Errorf("failed batching: %w", err) + break } - seenSpace[a.Space] = struct{}{} - seenAlias[a] = struct{}{} - vaHashKinds = append(vaHashKinds, hashKind) - vaHashes = append(vaHashes, hash) - vaSpaces = append(vaSpaces, a.Space.Value()) - vaNames = append(vaNames, a.Name) } - if vuln.Self.Valid() { - seenSpace[vuln.Self.Space] = struct{}{} - seenAlias[vuln.Self] = struct{}{} - vsHashKinds = append(vsHashKinds, hashKind) - vsHashes = append(vsHashes, hash) - vsSpaces = append(vsSpaces, vuln.Self.Space.Value()) - vsNames = append(vsNames, vuln.Self.Name) - } - - if ct := batch.Len(); ct < 1000 { - return true - } - if err = flush(); err != nil { - err = fmt.Errorf("failed batching: %w", err) - return false - } - return true - }) + } if err != nil { return uuid.Nil, fmt.Errorf("iterating on vulnerabilities: %w", err) } @@ -382,51 +383,6 @@ ON CONFLICT DO NOTHING;` updateVulnerabilitiesCounter.WithLabelValues("insert_batch", strconv.FormatBool(delta)).Add(1) updateVulnerabilitiesDuration.WithLabelValues("insert_batch", strconv.FormatBool(delta)).Observe(time.Since(start).Seconds()) - - // Insert alias namespaces and aliases outside the transaction to avoid - // deadlocks when concurrent updaters race to insert the same namespaces. - if len(seenSpace) > 0 { - spaces := make([]string, 0, len(seenSpace)) - for h := range seenSpace { - spaces = append(spaces, h.Value()) - } - aliasSpaces := make([]string, 0, len(seenAlias)) - aliasNames := make([]string, 0, len(seenAlias)) - for a := range seenAlias { - aliasSpaces = append(aliasSpaces, a.Space.Value()) - aliasNames = append(aliasNames, a.Name) - } - - conn, err := s.pool.Acquire(ctx) - if err != nil { - return uuid.Nil, fmt.Errorf("acquiring connection for aliases: %w", err) - } - defer conn.Release() - - if _, err := conn.Exec(ctx, insertAliasNamespaces, spaces); err != nil { - return uuid.Nil, fmt.Errorf("failed to insert alias namespaces: %w", err) - } - if _, err := conn.Exec(ctx, insertAliases, aliasSpaces, aliasNames); err != nil { - return uuid.Nil, fmt.Errorf("failed to insert aliases: %w", err) - } - } - - // Bulk-link aliases and self references. Two single statements replace the - // former per-vulnerability hash-lookup subqueries queued in the batch above. - start = time.Now() - if len(vaHashKinds) > 0 { - if _, err := tx.Exec(ctx, bulkLinkAliases, vaHashKinds, vaHashes, vaSpaces, vaNames); err != nil { - return uuid.Nil, fmt.Errorf("failed to bulk link vulnerability aliases: %w", err) - } - } - if len(vsHashKinds) > 0 { - if _, err := tx.Exec(ctx, bulkLinkSelf, vsHashKinds, vsHashes, vsSpaces, vsNames); err != nil { - return uuid.Nil, fmt.Errorf("failed to bulk link vulnerability self aliases: %w", err) - } - } - updateVulnerabilitiesCounter.WithLabelValues("link_aliases", strconv.FormatBool(delta)).Add(1) - updateVulnerabilitiesDuration.WithLabelValues("link_aliases", strconv.FormatBool(delta)).Observe(time.Since(start).Seconds()) - if err := tx.Commit(ctx); err != nil { return uuid.Nil, fmt.Errorf("failed to commit transaction: %w", err) } @@ -448,6 +404,13 @@ func skipVulnerability(v *claircore.Vulnerability) bool { return v.Package == nil || v.Package.Name == "" } +func rangekind(r *claircore.Range) (kind string) { + if r == nil || r.Lower.Kind != r.Upper.Kind { + return "" + } + return r.Lower.Kind +} + func rangefmt(r *claircore.Range) (kind *string, lower, upper string) { lower, upper = "{}", "{}" if r == nil || r.Lower.Kind != r.Upper.Kind {