From 90bd0389ef3f9547d503e2424c7ebc1db0be348b Mon Sep 17 00:00:00 2001 From: Martin Hutchinson Date: Thu, 24 Jul 2025 15:37:53 +0000 Subject: [PATCH 1/3] [VIndex] calculate and confirm input log hash This removes a lot of in-memory state, and creates a clearer invariant. The compact range now tracks the size of the ingested log, and keeps a summary of the hash. The latest verified checkpoint is now stored into a Pebble database. In the same database, we periodically store the latest tree size and compact range state too. This allows us to recover from any terminated runs, and pick up at the last index where we persisted the compact range. This meant updating the WAL so that control was inverted when constructing it; instead of the WAL saying what the index to start mapping from is, the mapping code gets this size from the compact range stored in Pebble, and then instructs that WAL to start appending after that size, truncating any dangling entries. This required fixing a couple of places that were implemented wrongly, but only came to light when confirming the tree root: - one of the tests wasn't hashing leaves properly - the logandmap binary was fetching too many leaves --- go.mod | 26 ++++ go.sum | 91 ++++++++++++++ vindex/cmd/logandmap/main.go | 14 ++- vindex/map.go | 238 ++++++++++++++++++++++++++++------- vindex/map_test.go | 10 +- vindex/wal.go | 106 +++++++++------- vindex/wal_test.go | 127 +++++++++++-------- 7 files changed, 466 insertions(+), 146 deletions(-) diff --git a/go.mod b/go.mod index 0bcdf9b..eb29320 100644 --- a/go.mod +++ b/go.mod @@ -15,14 +15,40 @@ require ( ) require ( + github.com/DataDog/zstd v1.4.5 // indirect + github.com/beorn7/perks v1.0.1 // indirect github.com/cenkalti/backoff/v5 v5.0.2 // indirect + github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/cockroachdb/errors v1.11.3 // indirect + github.com/cockroachdb/fifo v0.0.0-20240606204812-0bbfbd93a7ce // indirect + github.com/cockroachdb/logtags v0.0.0-20230118201751-21c54148d20b // indirect + github.com/cockroachdb/pebble v1.1.5 // indirect + github.com/cockroachdb/redact v1.1.5 // indirect + github.com/cockroachdb/tokenbucket v0.0.0-20230807174530-cc333fc44b06 // indirect + github.com/getsentry/sentry-go v0.27.0 // indirect github.com/go-logr/logr v1.4.3 // indirect github.com/go-logr/stdr v1.2.2 // indirect + github.com/gogo/protobuf v1.3.2 // indirect + github.com/golang/protobuf v1.5.3 // indirect + github.com/golang/snappy v0.0.4 // indirect github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect + github.com/klauspost/compress v1.18.0 // indirect + github.com/kr/pretty v0.3.1 // indirect + github.com/kr/text v0.2.0 // indirect + github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect + github.com/pkg/errors v0.9.1 // indirect + github.com/prometheus/client_golang v1.15.0 // indirect + github.com/prometheus/client_model v0.3.0 // indirect + github.com/prometheus/common v0.42.0 // indirect + github.com/prometheus/procfs v0.9.0 // indirect + github.com/rogpeppe/go-internal v1.13.1 // indirect go.opentelemetry.io/auto/sdk v1.1.0 // indirect go.opentelemetry.io/otel v1.37.0 // indirect go.opentelemetry.io/otel/metric v1.37.0 // indirect go.opentelemetry.io/otel/trace v1.37.0 // indirect golang.org/x/crypto v0.40.0 // indirect golang.org/x/exp v0.0.0-20240325151524-a685a6edb6d8 // indirect + golang.org/x/sys v0.34.0 // indirect + golang.org/x/text v0.27.0 // indirect + google.golang.org/protobuf v1.36.6 // indirect ) diff --git a/go.sum b/go.sum index 7d91588..9509a01 100644 --- a/go.sum +++ b/go.sum @@ -1,24 +1,79 @@ filippo.io/torchwood v0.5.1-0.20250605130057-fa65d721a6ce h1:8bVOhkZ6uQn3qCmrz/F7Rz0epygH9KO3dgbS9g/zclk= filippo.io/torchwood v0.5.1-0.20250605130057-fa65d721a6ce/go.mod h1:n82QaQC2EzvctEMkM1KAahZaAcQ2uZoCdn051sUXOQY= +github.com/DataDog/zstd v1.4.5 h1:EndNeuB0l9syBZhut0wns3gV1hL8zX8LIu6ZiVHWLIQ= +github.com/DataDog/zstd v1.4.5/go.mod h1:1jcaCB/ufaK+sKp1NBhlGmpz41jOoPQ35bpF36t7BBo= +github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= +github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= github.com/cenkalti/backoff/v5 v5.0.2 h1:rIfFVxEf1QsI7E1ZHfp/B4DF/6QBAUhmgkxc0H7Zss8= github.com/cenkalti/backoff/v5 v5.0.2/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F97BxZthm/crw= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/cockroachdb/errors v1.11.3 h1:5bA+k2Y6r+oz/6Z/RFlNeVCesGARKuC6YymtcDrbC/I= +github.com/cockroachdb/errors v1.11.3/go.mod h1:m4UIW4CDjx+R5cybPsNrRbreomiFqt8o1h1wUVazSd8= +github.com/cockroachdb/fifo v0.0.0-20240606204812-0bbfbd93a7ce h1:giXvy4KSc/6g/esnpM7Geqxka4WSqI1SZc7sMJFd3y4= +github.com/cockroachdb/fifo v0.0.0-20240606204812-0bbfbd93a7ce/go.mod h1:9/y3cnZ5GKakj/H4y9r9GTjCvAFta7KLgSHPJJYc52M= +github.com/cockroachdb/logtags v0.0.0-20230118201751-21c54148d20b h1:r6VH0faHjZeQy818SGhaone5OnYfxFR/+AzdY3sf5aE= +github.com/cockroachdb/logtags v0.0.0-20230118201751-21c54148d20b/go.mod h1:Vz9DsVWQQhf3vs21MhPMZpMGSht7O/2vFW2xusFUVOs= +github.com/cockroachdb/pebble v1.1.5 h1:5AAWCBWbat0uE0blr8qzufZP5tBjkRyy/jWe1QWLnvw= +github.com/cockroachdb/pebble v1.1.5/go.mod h1:17wO9el1YEigxkP/YtV8NtCivQDgoCyBg5c4VR/eOWo= +github.com/cockroachdb/redact v1.1.5 h1:u1PMllDkdFfPWaNGMyLD1+so+aq3uUItthCFqzwPJ30= +github.com/cockroachdb/redact v1.1.5/go.mod h1:BVNblN9mBWFyMyqK1k3AAiSxhvhfK2oOZZ2lK+dpvRg= +github.com/cockroachdb/tokenbucket v0.0.0-20230807174530-cc333fc44b06 h1:zuQyyAKVxetITBuuhv3BI9cMrmStnpT18zmgmTxunpo= +github.com/cockroachdb/tokenbucket v0.0.0-20230807174530-cc333fc44b06/go.mod h1:7nc4anLGjupUW/PeY5qiNYsdNXj7zopG+eqsS7To5IQ= +github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/getsentry/sentry-go v0.27.0 h1:Pv98CIbtB3LkMWmXi4Joa5OOcwbmnX88sF5qbK3r3Ps= +github.com/getsentry/sentry-go v0.27.0/go.mod h1:lc76E2QywIyW8WuBnwl8Lc4bkmQH4+w1gwTf25trprY= github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= +github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q= +github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q= +github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U= +github.com/golang/protobuf v1.3.5/go.mod h1:6O5/vntMXwX2lRkT1hjjk0nAC1IDOTvTlVgjlRvqsdk= +github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk= +github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg= +github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY= +github.com/golang/snappy v0.0.4 h1:yAGX7huGHXlcLOEtBnF4w7FQwA26wojNCwOYAEhLjQM= +github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= +github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/gorilla/mux v1.8.1 h1:TuBL49tXwgrFYWhqrNgrUNEY92u81SPhu7sTdzQEiWY= github.com/gorilla/mux v1.8.1/go.mod h1:AKf9I4AEqPTmMytcMc0KkNouC66V3BtZ4qD5fmWSiMQ= github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k= github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM= +github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8= +github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= +github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo= +github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ= github.com/klauspost/cpuid/v2 v2.0.9 h1:lgaqFMSdTdQYdZ04uHyN2d/eKdOMyi2YLSvlQIBFYa4= github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg= +github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= +github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/matttproud/golang_protobuf_extensions v1.0.4 h1:mmDVorXM7PCGKw94cs5zkfA9PSy5pEvNWRP0ET0TIVo= +github.com/matttproud/golang_protobuf_extensions v1.0.4/go.mod h1:BSXmuO+STAnVfrANrmjBb36TMTDstsz7MSK+HVaYKv4= +github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA= +github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= +github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/prometheus/client_golang v1.15.0 h1:5fCgGYogn0hFdhyhLbw7hEsWxufKtY9klyvdNfFlFhM= +github.com/prometheus/client_golang v1.15.0/go.mod h1:e9yaBhRPU2pPNsZwE+JdQl0KEt1N9XgF6zxWmaC0xOk= +github.com/prometheus/client_model v0.3.0 h1:UBgGFHqYdG/TPFD1B1ogZywDqEkwp3fBMvqdiQ7Xew4= +github.com/prometheus/client_model v0.3.0/go.mod h1:LDGWKZIo7rky3hgvBe+caln+Dr3dPggB5dvjtD7w9+w= +github.com/prometheus/common v0.42.0 h1:EKsfXEYo4JpWMHH5cg+KOUWeuJSov1Id8zGR8eeI1YM= +github.com/prometheus/common v0.42.0/go.mod h1:xBwqVerjNdUDjgODMpudtOMwlOwf2SaTr1yjz4b7Zbc= +github.com/prometheus/procfs v0.9.0 h1:wzCHvIvM5SxWqYvwgVL7yJY8Lz3PKn49KQtpgMYJfhI= +github.com/prometheus/procfs v0.9.0/go.mod h1:+pB4zwohETzFnmlpe6yd2lSc+0/46IYZRB/chUwxUZY= +github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs= +github.com/rogpeppe/go-internal v1.13.1 h1:KvO1DLK/DRN07sQ1LQKScxyZJuNnedQ5/wKSR38lUII= +github.com/rogpeppe/go-internal v1.13.1/go.mod h1:uMEvuHeurkdAXX61udpOXGD/AzZDWNMNyH2VO9fmH0o= github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA= github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= github.com/transparency-dev/formats v0.0.0-20250723101439-be3b1008ec3a h1:l1RrmDw9xrVN/lbW/rzPJhjQ+dmsqOyItES9Ku/njFA= @@ -27,6 +82,8 @@ github.com/transparency-dev/merkle v0.0.2 h1:Q9nBoQcZcgPamMkGn7ghV8XiTZ/kRxn1yCG github.com/transparency-dev/merkle v0.0.2/go.mod h1:pqSy+OXefQ1EDUVmAJ8MUhHB9TXGuzVAT58PqBoHz1A= github.com/transparency-dev/tessera v0.2.1-0.20250722085756-7303218c6614 h1:EfxzWae/zdnfVA44jFIKohvd2JWlcanRzRYb1A/uiDU= github.com/transparency-dev/tessera v0.2.1-0.20250722085756-7303218c6614/go.mod h1:ilpKqGrwDD/6uop5nDj/X60o0qt33GK1uInjIcfZTP0= +github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= +github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA= go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A= go.opentelemetry.io/otel v1.37.0 h1:9zhNfelUvx0KBfu/gb+ZgeAfAgtWrfHJZcAqFC228wQ= @@ -35,14 +92,48 @@ go.opentelemetry.io/otel/metric v1.37.0 h1:mvwbQS5m0tbmqML4NqK+e3aDiO02vsf/Wgbsd go.opentelemetry.io/otel/metric v1.37.0/go.mod h1:04wGrZurHYKOc+RKeye86GwKiTb9FKm1WHtO+4EVr2E= go.opentelemetry.io/otel/trace v1.37.0 h1:HLdcFNbRQBE2imdSEgm/kwqmQj1Or1l/7bW6mxVK7z4= go.opentelemetry.io/otel/trace v1.37.0/go.mod h1:TlgrlQ+PtQO5XFerSPUYG0JSgGyryXewPGyayAWSBS0= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= +golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= golang.org/x/crypto v0.40.0 h1:r4x+VvoG5Fm+eJcxMaY8CQM7Lb0l1lsmjGBQ6s8BfKM= golang.org/x/crypto v0.40.0/go.mod h1:Qr1vMER5WyS2dfPHAlsOj01wgLbsyWtFn/aY+5+ZdxY= golang.org/x/exp v0.0.0-20240325151524-a685a6edb6d8 h1:aAcj0Da7eBAtrTp03QXWvm88pSyOt+UgdZw2BFZ+lEw= golang.org/x/exp v0.0.0-20240325151524-a685a6edb6d8/go.mod h1:CQ1k9gNrJ50XIzaKCRR2hssIjF07kZFEiieALBM/ARQ= +golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= +golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.26.0 h1:EGMPT//Ezu+ylkCijjPc+f4Aih7sZvaAr+O3EHBxvZg= golang.org/x/mod v0.26.0/go.mod h1:/j6NAhSk8iQ723BGAUyoAcn7SlD7s15Dp9Nd/SfeaFQ= +golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= +golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= +golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.16.0 h1:ycBJEhp9p4vXvUZNszeOq0kGTPghopOL8q0fq3vstxw= golang.org/x/sync v0.16.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA= +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.34.0 h1:H5Y5sJ2L2JRdyv7ROF1he/lPdvFsd0mJHFw2ThKHxLA= +golang.org/x/sys v0.34.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= +golang.org/x/text v0.27.0 h1:4fGWRpyh641NLlecmyl4LOe6yDdfaYNrGb2zdfo4JV4= +golang.org/x/text v0.27.0/go.mod h1:1D28KMCvyooCX9hBiosv5Tz/+YLxj0j7XhWjpSUF7CU= +golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= +golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= +golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= +golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= +golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw= +google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc= +google.golang.org/protobuf v1.36.6 h1:z1NpPI8ku2WgiWnf+t9wTPsn6eP1L7ksHUlkfLvd9xY= +google.golang.org/protobuf v1.36.6/go.mod h1:jduwjTPXsFjZGTmRluh+L6NjiWu7pchiJ2/5YcXBHnY= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= k8s.io/klog/v2 v2.130.1 h1:n9Xl7H1Xvksem4KFG4PYbdQCQxqc/tTUyrgXaOhHSzk= diff --git a/vindex/cmd/logandmap/main.go b/vindex/cmd/logandmap/main.go index 640bb13..6c41c12 100644 --- a/vindex/cmd/logandmap/main.go +++ b/vindex/cmd/logandmap/main.go @@ -78,11 +78,14 @@ func run(ctx context.Context) error { return errors.New("storage_dir must be set") } inputLogDir := path.Join(*storageDir, "inputlog") - walPath := path.Join(*storageDir, "index.wal") + mapRoot := path.Join(*storageDir, "vindex") if err := os.MkdirAll(inputLogDir, 0o755); err != nil { return fmt.Errorf("failed to create input log directory: %v", err) } + if err := os.MkdirAll(mapRoot, 0o755); err != nil { + return fmt.Errorf("failed to create vindex directory: %v", err) + } // Gather the info needed for reading/writing checkpoints ils, ilv := getInputLogSignerVerifierOrDie() @@ -95,7 +98,7 @@ func run(ctx context.Context) error { inputAppender, inputShutdown, inputReader, err := tessera.NewAppender(ctx, ild, tessera.NewAppendOptions(). WithCheckpointSigner(ils). - WithCheckpointInterval(10*time.Second). + WithCheckpointInterval(5*time.Second). WithBatching(256, time.Second)) if err != nil { return fmt.Errorf("failed to get appender: %v", err) @@ -113,7 +116,7 @@ func run(ctx context.Context) error { cp, _, _, err := log.ParseCheckpoint(cpRaw, ilv.Name(), ilv) return cp, err } - vi, err := vindex.NewVerifiableIndex(ctx, inputLog, inputLogCpParseFn, mapFnFromFlags(), walPath) + vi, err := vindex.NewVerifiableIndex(ctx, inputLog, inputLogCpParseFn, mapFnFromFlags(), mapRoot) if err != nil { return fmt.Errorf("failed to create vindex: %v", err) } @@ -140,7 +143,10 @@ func (s logReaderSource) Checkpoint(ctx context.Context) (checkpoint []byte, err } func (s logReaderSource) Leaves(ctx context.Context, start, end uint64) iter.Seq2[[]byte, error] { - bi := client.EntryBundles(ctx, 2, s.r.IntegratedSize, s.r.ReadEntryBundle, start, end-start) + tsf := func(ctx context.Context) (uint64, error) { + return end, nil + } + bi := client.EntryBundles(ctx, 2, tsf, s.r.ReadEntryBundle, start, end-start) unbundleFn := func(bundle []byte) ([][]byte, error) { eb := &api.EntryBundle{} if err := eb.UnmarshalText(bundle); err != nil { diff --git a/vindex/map.go b/vindex/map.go index b95b4f1..72ed9c7 100644 --- a/vindex/map.go +++ b/vindex/map.go @@ -19,21 +19,34 @@ package vindex import ( + "bytes" "context" "crypto/sha256" + "encoding/base64" "encoding/binary" "fmt" "io" "iter" + "os" + "path" + "strconv" "sync" "time" "filippo.io/torchwood/mpt" + "github.com/cockroachdb/pebble" "github.com/transparency-dev/formats/log" - "golang.org/x/sync/errgroup" + "github.com/transparency-dev/merkle/compact" + "github.com/transparency-dev/merkle/rfc6962" "k8s.io/klog/v2" ) +const ( + db_latestCheckpointKey = "latestCheckpoint" + db_compactRangeKey = "compactRange" + db_compactRangeSizeKey = "compactRangeSize" +) + // MapFn takes the raw leaf data from a log entry and outputs the SHA256 hashes // of the keys at which this leaf should be indexed under. // A leaf can be recorded at any number of entries, including no entries (in which case an empty slice must be returned). @@ -65,8 +78,59 @@ type OpenCheckpointFn func(cpRaw []byte) (*log.Checkpoint, error) // path. // Note that only one IndexBuilder should exist for any given walPath at any time. The behaviour is unspecified, // but likely broken, if multiple processes are writing to the same file at any given time. -func NewVerifiableIndex(ctx context.Context, inputLog InputLog, inputLogParseFn OpenCheckpointFn, mapFn MapFn, walPath string) (*VerifiableIndex, error) { - wal, ws, err := newWalWriter(walPath) +func NewVerifiableIndex(ctx context.Context, inputLog InputLog, inputLogParseFn OpenCheckpointFn, mapFn MapFn, rootDir string) (*VerifiableIndex, error) { + stateDir := path.Join(rootDir, "state") + if err := os.MkdirAll(stateDir, 0o755); err != nil { + return nil, err + } + db, err := pebble.Open(stateDir, &pebble.Options{}) + if err != nil { + return nil, fmt.Errorf("pebble.Open(): %v", err) + } + + // Helpful wrapper to convert Closer to something that can be safely deferred (according to the linter) + logClose := func(c io.Closer) { + if err := c.Close(); err != nil { + klog.Error(err) + } + } + + // Load the compact range we have calculated so far, and the size persisted. We MUST start + // from this index in order to have properly verified the state of the input log. + crf := compact.RangeFactory{Hash: rfc6962.DefaultHasher.HashChildren} + var cr *compact.Range + var size uint64 + snap := db.NewSnapshot() + defer logClose(snap) + + if sizeBs, sizeCloser, err := snap.Get([]byte(db_compactRangeSizeKey)); err != nil { + if err != pebble.ErrNotFound { + return nil, fmt.Errorf("pebble.Get(): %v", err) + } + size = 0 + cr = crf.NewEmptyRange(0) + } else { + crBs, crCloser, err := snap.Get([]byte(db_compactRangeKey)) + if err != nil { + return nil, fmt.Errorf("pebble.Get(): %v", err) + } + defer logClose(crCloser) + defer logClose(sizeCloser) + + size = binary.BigEndian.Uint64(sizeBs) + crHashes := make([][]byte, len(crBs)/sha256.Size) + for i := range crHashes { + crHashes[i] = crBs[i*sha256.Size : (i+1)*sha256.Size] + } + klog.Infof("Loaded state: size=%d, hashes=%d", size, len(crHashes)) + cr, err = crf.NewRange(0, size, crHashes) + if err != nil { + return nil, fmt.Errorf("NewRange: %v", err) + } + } + + walPath := path.Join(rootDir, "map.wal") + wal, err := newWalWriter(walPath, size) if err != nil { return nil, err } @@ -83,15 +147,17 @@ func NewVerifiableIndex(ctx context.Context, inputLog InputLog, inputLogParseFn inputLogParseFn: inputLogParseFn, mapFn: mapFn, walWriter: wal, - nextIndex: ws, + db: db, + r: cr, } b := &VerifiableIndex{ mapper: mapper, walReader: reader, + db: db, vindex: *mpt.NewTree(sha256.Sum256, vtreeStorage), data: map[[32]byte][]uint64{}, } - if err := b.buildMap(ctx, ws); err != nil { + if err := b.buildMap(ctx); err != nil { return nil, fmt.Errorf("failed to build map: %v", err) } return b, nil @@ -104,36 +170,15 @@ type inputLogMapper struct { inputLogParseFn OpenCheckpointFn mapFn MapFn walWriter *walWriter + db *pebble.DB - nextIndex uint64 // nextIndex is the next index in the log to consume - inputLogCpSize uint64 // cpSize is the tree size of rawCp. Used to sync on WAL. + r *compact.Range } func (m *inputLogMapper) close() error { return m.walWriter.close() } -// available returns whether this is work to do. -// TODO(mhutchinson): this can probably be deleted -func (m *inputLogMapper) available(ctx context.Context) bool { - rawCp, err := m.inputLog.Checkpoint(ctx) - if err != nil { - klog.Warningf("Failed to get latest checkpoint from DB: %s", err) - return false - } - cp, err := m.inputLogParseFn(rawCp) - if err != nil { - klog.Warningf("Failed to parse checkpoint: %s", err) - return false - } - - // TODO(mhutchinson): remove this and replace with disk persistence? - defer func() { - m.inputLogCpSize = cp.Size - }() - return cp.Size > m.inputLogCpSize -} - // syncFromInputLog reads the latest checkpoint from the input log, and ensures that the WAL // contains a corresponding entry for every index committed to by that checkpoint. // @@ -142,14 +187,30 @@ func (m *inputLogMapper) available(ctx context.Context) bool { // cloneDB, which performed this validation. Implementing this will require the index to store some // state alongside the WAL which contains a compact range of its current progress. func (m *inputLogMapper) syncFromInputLog(ctx context.Context) error { - if m.inputLogCpSize > m.nextIndex { + rawCp, err := m.inputLog.Checkpoint(ctx) + if err != nil { + return fmt.Errorf("failed to get latest checkpoint from DB: %s", err) + } + cp, err := m.inputLogParseFn(rawCp) + if err != nil { + return fmt.Errorf("failed to parse checkpoint: %s", err) + } + + if m.r.End() < cp.Size { ctx, done := context.WithCancel(ctx) defer done() - for l, err := range m.inputLog.Leaves(ctx, m.nextIndex, m.inputLogCpSize) { - idx := m.nextIndex + for l, err := range m.inputLog.Leaves(ctx, m.r.End(), cp.Size) { + idx := m.r.End() if err != nil { return fmt.Errorf("failed to read leaf at index %d: %v", idx, err) } + if idx >= cp.Size { + return fmt.Errorf("expected stop at cp.Size=%d, but got leaf at index=%d", cp.Size, idx) + } + + if err := m.r.Append(rfc6962.DefaultHasher.HashLeaf(l), nil); err != nil { + return fmt.Errorf("failed to update compact range: %v", err) + } // Apply the MapFn in as safe a way as possible. This involves trapping any panics // and failing gracefully. @@ -166,8 +227,9 @@ func (m *inputLogMapper) syncFromInputLog(ctx context.Context) error { if mapErr != nil { return mapErr } - m.nextIndex++ - if len(hashes) == 0 && idx < m.inputLogCpSize-1 { + + storeCompactRange := m.r.End()%256 == 0 || m.r.End() == cp.Size + if len(hashes) == 0 && !storeCompactRange { // We can skip writing out values with no hashes, as long as we're // not at the end of the log. // If we are at the end of the log, we need to write out a value as a sentinel @@ -177,8 +239,54 @@ func (m *inputLogMapper) syncFromInputLog(ctx context.Context) error { if err := m.walWriter.append(idx, hashes); err != nil { return fmt.Errorf("failed to add index to entry for leaf %d: %v", idx, err) } + if storeCompactRange { + // Periodically store the validated compact range consumed so far. + // The choice to align with every 256 entries is an implicit bias towards + // supporting tlog-tiles. + if err := m.walWriter.flush(); err != nil { + return fmt.Errorf("failed to flush the WAL: %v", err) + } + if err := m.storeState(); err != nil { + return fmt.Errorf("failed to store incremental state: %v", err) + } + } } } + if err := m.walWriter.flush(); err != nil { + return fmt.Errorf("failed to flush: %v", err) + } + + // Calculate the root hash, and if it checks out, store the checkpoint to indicate safety. + hash, err := m.r.GetRootHash(nil) + if err != nil { + return fmt.Errorf("failed to get root hash from compact range: %v", err) + } + if !bytes.Equal(hash, cp.Hash) { + return fmt.Errorf("calculated hash for tree size %d is %x, but checkpoint commits to %x", cp.Size, hash, cp.Hash) + } + if err := m.db.Set([]byte(db_latestCheckpointKey), rawCp, pebble.Sync); err != nil { + return fmt.Errorf("failed to update state: %v", err) + } + + klog.Infof("synced WAL to size %d", cp.Size) + return nil +} + +func (m *inputLogMapper) storeState() error { + flatSlice := make([]byte, len(m.r.Hashes())*32) + for i, arr := range m.r.Hashes() { + copy(flatSlice[i*32:], arr[:]) + } + b := m.db.NewBatch() + if err := b.Set([]byte(db_compactRangeKey), flatSlice, pebble.Sync); err != nil { + return fmt.Errorf("failed to update state: %v", err) + } + if err := b.Set([]byte(db_compactRangeSizeKey), binary.BigEndian.AppendUint64(nil, m.r.End()), pebble.Sync); err != nil { + return fmt.Errorf("failed to update state: %v", err) + } + if err := m.db.Apply(b, pebble.Sync); err != nil { + return fmt.Errorf("failed to update state: %v", err) + } return nil } @@ -187,6 +295,7 @@ func (m *inputLogMapper) syncFromInputLog(ctx context.Context) error { type VerifiableIndex struct { mapper *inputLogMapper walReader *walReader + db *pebble.DB indexMu sync.RWMutex // covers vindex and data vindex mpt.Tree @@ -231,27 +340,36 @@ func (b *VerifiableIndex) Lookup(key [sha256.Size]byte) (indices []uint64, size // Update checks the input log for a new Checkpoint, and ensures that the Verifiable Index // is updated to the corresponding size. func (b *VerifiableIndex) Update(ctx context.Context) error { - if !b.mapper.available(ctx) { - return nil + // TODO(mhutchinson): look for options to improve concurrency again here. + if err := b.mapper.syncFromInputLog(ctx); err != nil { + return err } - - newSize := b.mapper.inputLogCpSize - eg, cctx := errgroup.WithContext(ctx) - eg.Go(func() error { return b.mapper.syncFromInputLog(cctx) }) - eg.Go(func() error { return b.buildMap(cctx, newSize) }) - - err := eg.Wait() - - return err + return b.buildMap(ctx) } // buildMap reads from the WAL until the file has been consumed and the map has been // built up the provided size. -func (b *VerifiableIndex) buildMap(ctx context.Context, toSize uint64) error { +func (b *VerifiableIndex) buildMap(ctx context.Context) error { startWal := time.Now() updatedKeys := make(map[[32]byte]bool) // Allows us to efficiently update vindex after first init - for i := b.servingSize; i < toSize; { + cpRaw, closer, err := b.db.Get([]byte(db_latestCheckpointKey)) + if err != nil { + if err == pebble.ErrNotFound { + // If the key isn't there then nothing to do. + return nil + } + return fmt.Errorf("failed to read latest checkpoint: %v", err) + } + if err := closer.Close(); err != nil { + return fmt.Errorf("failed to close: %v", err) + } + _, size, _, err := checkpointUnsafe(cpRaw) + if err != nil { + return fmt.Errorf("failed to parse checkpoint: %v", err) + } + + for i := b.servingSize; i < size; { select { case <-ctx.Done(): return ctx.Err() @@ -320,7 +438,33 @@ func (b *VerifiableIndex) buildMap(ctx context.Context, toSize uint64) error { durationVIndex := time.Since(startVIndex) durationTotal := time.Since(startWal) - b.servingSize = toSize + b.servingSize = size klog.Infof("buildMap: total=%s (wal=%s, vindex=%s)", durationTotal, durationWal, durationVIndex) return nil } + +// checkpointUnsafe parses a checkpoint without performing any signature verification. +// This is intended to be as fast as possible, but sacrifices safety because it skips verifying +// the note signature. +// +// Parsing a checkpoint like this is only acceptable in a process where the checkpoint has already +// been verified properly, and hasn't left the TCB since being checked. In this code, the on-disk +// storage is considered to be in the TCB, and thus we can skip fully verify it a second time. +func checkpointUnsafe(rawCp []byte) (string, uint64, []byte, error) { + parts := bytes.SplitN(rawCp, []byte{'\n'}, 4) + if want, got := 4, len(parts); want != got { + return "", 0, nil, fmt.Errorf("invalid checkpoint: %q", rawCp) + } + origin := string(parts[0]) + sizeStr := string(parts[1]) + hashStr := string(parts[2]) + size, err := strconv.ParseUint(sizeStr, 10, 64) + if err != nil { + return "", 0, nil, fmt.Errorf("failed to turn checkpoint size of %q into uint64: %v", sizeStr, err) + } + hash, err := base64.StdEncoding.DecodeString(hashStr) + if err != nil { + return "", 0, nil, fmt.Errorf("failed to decode hash: %v", err) + } + return origin, size, hash, nil +} diff --git a/vindex/map_test.go b/vindex/map_test.go index 8262ee5..1c6074d 100644 --- a/vindex/map_test.go +++ b/vindex/map_test.go @@ -64,13 +64,19 @@ func TestVerifiableIndex(t *testing.T) { } return [][32]byte{sha256.Sum256(key)} } - f, err := os.CreateTemp("", "testWal") + f, err := os.CreateTemp("", "vindexTestDir") if err != nil { t.Fatal(err) } if err := f.Close(); err != nil { t.Fatal(err) } + if err := os.Remove(f.Name()); err != nil { + t.Fatal(err) + } + if err := os.MkdirAll(f.Name(), 0o755); err != nil { + t.Fatal(err) + } vi, err := NewVerifiableIndex(ctx, inputLog, inputLogCpParseFn, mapFn, f.Name()) if err != nil { t.Fatal(err) @@ -137,7 +143,7 @@ func (s *inMemoryTreeSource) Leaves(ctx context.Context, start, end uint64) iter func (s *inMemoryTreeSource) Append(leafStr string) { leaf := []byte(leafStr) s.leaves = append(s.leaves, leaf) - s.t.Append(leaf) + s.t.Append(rfc6962.DefaultHasher.HashLeaf(leaf)) } func mustHashEncode(data string) string { diff --git a/vindex/wal.go b/vindex/wal.go index 7ad3dd2..d4b505e 100644 --- a/vindex/wal.go +++ b/vindex/wal.go @@ -27,46 +27,44 @@ import ( "k8s.io/klog/v2" ) -func newWalWriter(walPath string) (*walWriter, uint64, error) { +// newWalWriter creates a WAL writer that uses the file at the given path. +// This will verify that the tail of the WAL file is well-formed, and truncate the file +// if there is any trailing corruption. It will ensure that the file is truncated so that +// the last index there is treeSize-1. This allows the WAL to be +// flushed ahead of storing state about the input log merkle tree, and then slightly +// truncated if the WAL ends up being ahead due to process termination. +// If no entry for treeSize-1 can be found, then this returns an error. +func newWalWriter(walPath string, treeSize uint64) (*walWriter, error) { w := &walWriter{ walPath: walPath, } - idx, err := w.init() - return w, idx, err -} - -// walWriter provides the methods needed by the processor of the Input Log when interacting -// with the WAL. init provides the index that this processor should start from, and append -// allows new mapped entries to be added to the WAL. -type walWriter struct { - walPath string - f *os.File -} - -// init verifies that the log is in good shape, and returns the index that is expected next. -// It also opens the log for appending to. -// -// Note that it returns the next expected index to avoid awkwardness with the meaning of 0, -// which could mean 0 was successfully read from a previous run, or that there was no log. -func (l *walWriter) init() (uint64, error) { ffs := os.O_WRONLY | os.O_APPEND - idx, err := validate(l.walPath) + var lastIndex uint64 + if treeSize > 0 { + lastIndex = treeSize - 1 + } + err := validate(walPath, lastIndex) if err != nil { if !errors.Is(err, os.ErrNotExist) { - return idx, err + return nil, err } ffs |= os.O_CREATE | os.O_EXCL - } else { - // If the file exists, then we expect the next index to be returned - idx++ } // Open the file for writing in append-only, creating it if needed - l.f, err = os.OpenFile(l.walPath, ffs, 0o644) + w.f, err = os.OpenFile(walPath, ffs, 0o644) if err != nil { - return 0, fmt.Errorf("failed to open file for writing: %s", err) + return nil, fmt.Errorf("failed to open file for writing: %s", err) } - return idx, err + return w, err +} + +// walWriter provides the methods needed by the processor of the Input Log when interacting +// with the WAL. init provides the index that this processor should start from, and append +// allows new mapped entries to be added to the WAL. +type walWriter struct { + walPath string + f *os.File } func (l *walWriter) close() error { @@ -76,26 +74,26 @@ func (l *walWriter) close() error { // validate reads the file and determines what the last mapped log index was, and returns it. // The assumption is that all lines ending with a newline were written correctly. // If there are any errors in the file then this throws an error. -func validate(walPath string) (uint64, error) { +func validate(walPath string, lastIdx uint64) error { f, err := os.OpenFile(walPath, os.O_RDWR, 0o644) if err != nil { - return 0, err + return err } defer func() { _ = f.Close() }() fi, err := f.Stat() if err != nil { - return 0, err + return err } // Handle trivial case of empty file - size := fi.Size() - if size == 0 { + fileSize := fi.Size() + if fileSize == 0 { if err := os.Remove(walPath); err != nil { - return 0, fmt.Errorf("failed to delete empty file: %s", err) + return fmt.Errorf("failed to delete empty file: %s", err) } - return 0, os.ErrNotExist + return os.ErrNotExist } // Read from the end of the file in stripes, terminating when we either: @@ -104,7 +102,7 @@ func validate(walPath string) (uint64, error) { var buffer string const stripeSize = 1024 readStripe := make([]byte, stripeSize) - seekPos := size - stripeSize + seekPos := fileSize - stripeSize droppedTail := false for { @@ -115,35 +113,55 @@ func validate(walPath string) (uint64, error) { seekPos = 0 } if _, err := f.ReadAt(readStripe, seekPos); err != nil { - return 0, err + return err } buffer = string(readStripe) + buffer + var truncFrom int64 = -1 + defer func() { + if truncFrom >= 0 { + klog.Warningf("Dropping trailing %d bytes from WAL", fileSize-truncFrom) + if err := f.Truncate(truncFrom); err != nil { + panic(fmt.Errorf("failed to truncate WAL: %v", err)) + } + } + }() + for i := strings.LastIndex(buffer, "\n"); i > 0; i = strings.LastIndex(buffer, "\n") { p := buffer[i+1:] buffer = buffer[:i] if !droppedTail { droppedTail = true if len(p) > 0 { - truncPos := seekPos + int64(i) + 1 - klog.Warningf("Dropping tail part from WAL: %q", p) - if err := f.Truncate(truncPos); err != nil { - return 0, fmt.Errorf("failed to truncate WAL: %v", err) - } + truncFrom = seekPos + int64(i) + 1 } continue } idx, _, err := unmarshalWalEntry(p) - return idx, err + if err != nil { + return err + } + switch { + case idx > lastIdx: + truncFrom = seekPos + int64(i) + 1 + case idx == lastIdx: + return nil + case idx < lastIdx: + return fmt.Errorf("failed to find index %d (terminated after rewinding to %d)", lastIdx, idx) + } } if seekPos == 0 { - idx, _, err := unmarshalWalEntry(buffer) - return idx, err + _, _, err := unmarshalWalEntry(buffer) + return err } seekPos = seekPos - stripeSize } } +func (l *walWriter) flush() error { + return l.f.Sync() +} + func (l *walWriter) append(idx uint64, hashes [][32]byte) error { e, err := marshalWalEntry(idx, hashes) if err != nil { diff --git a/vindex/wal_test.go b/vindex/wal_test.go index 17d60b1..7d050e0 100644 --- a/vindex/wal_test.go +++ b/vindex/wal_test.go @@ -82,19 +82,16 @@ func TestWriteAheadLog_init(t *testing.T) { if err := f.Close(); err != nil { t.Fatal(err) } - wal, idx, err := newWalWriter(f.Name()) - defer func() { - _ = wal.close() - }() + wal, err := newWalWriter(f.Name(), 0) if gotErr := err != nil; gotErr != tC.wantErr { t.Fatalf("wantErr != gotErr (%t != %t) %v", tC.wantErr, gotErr, err) } if tC.wantErr { return } - if idx != tC.wantIdx { - t.Errorf("want idx %v but got %v", tC.wantIdx, idx) - } + defer func() { + _ = wal.close() + }() }) } } @@ -110,16 +107,13 @@ func TestWriteAheadLog_truncate(t *testing.T) { if err := f.Close(); err != nil { t.Fatal(err) } - wal, idx, err := newWalWriter(f.Name()) + wal, err := newWalWriter(f.Name(), 3) if err != nil { t.Fatal(err) } defer func() { _ = wal.close() }() - if got, want := idx, uint64(3); got != want { - t.Errorf("expected next index %d, but got %d", want, got) - } contents, err := os.ReadFile(f.Name()) if err != nil { @@ -131,46 +125,84 @@ func TestWriteAheadLog_truncate(t *testing.T) { } func TestWriteAheadLog_roundtrip(t *testing.T) { - f, err := os.CreateTemp("", "testWal") - if err != nil { - t.Fatal(err) - } - if err := f.Close(); err != nil { - t.Fatal(err) - } - if err := os.Remove(f.Name()); err != nil { - t.Fatal(err) + testCases := []struct { + desc string + entriesToWrite uint64 + treeSizeNeeded uint64 + wantErr bool + }{ + { + desc: "write 2, read 2", + entriesToWrite: 2, + treeSizeNeeded: 2, + wantErr: false, + }, { + desc: "write 200, read 200", + entriesToWrite: 200, + treeSizeNeeded: 200, + wantErr: false, + }, { + desc: "write 50, read 20", + entriesToWrite: 50, + treeSizeNeeded: 20, + wantErr: false, + }, } + for _, tC := range testCases { + t.Run(tC.desc, func(t *testing.T) { + f, err := os.CreateTemp("", "testWal") + if err != nil { + t.Fatal(err) + } + if err := f.Close(); err != nil { + t.Fatal(err) + } + if err := os.Remove(f.Name()); err != nil { + t.Fatal(err) + } - wal, idx, err := newWalWriter(f.Name()) - if err != nil { - t.Fatal(err) - } - if got, want := idx, uint64(0); got != want { - t.Fatalf("expected index %d, got %d", want, got) - } + wal, err := newWalWriter(f.Name(), 0) + if err != nil { + t.Fatal(err) + } - for i := range 33 { - hash := sha256.Sum256([]byte{byte(i)}) - if err := wal.append(uint64(i), [][32]byte{hash}); err != nil { - t.Error(err) - } - } + for i := range tC.entriesToWrite { + hash := sha256.Sum256([]byte{byte(i)}) + if err := wal.append(uint64(i), [][32]byte{hash}); err != nil { + t.Error(err) + } + } - if err := wal.close(); err != nil { - t.Error(err) - } + if err := wal.close(); err != nil { + t.Error(err) + } - wal, idx, err = newWalWriter(f.Name()) - if err != nil { - t.Fatal(err) - } - if got, want := idx, uint64(33); got != want { - t.Fatalf("expected index %d, got %d", want, got) - } + wal, err = newWalWriter(f.Name(), tC.treeSizeNeeded) + if err != nil { + t.Fatal(err) + } - if err := wal.close(); err != nil { - t.Error(err) + if err := wal.close(); err != nil { + t.Error(err) + } + + wr, err := newWalReader(f.Name()) + if err != nil { + t.Fatal(err) + } + + var lastIdx uint64 + for { + if idx, _, err := wr.next(); err != nil { + break + } else { + lastIdx = idx + } + } + if got, want := lastIdx, tC.treeSizeNeeded-1; got != want { + t.Errorf("expected reader to have last index of %d, but found %d", want, got) + } + }) } } @@ -186,13 +218,10 @@ func TestWriteAndWriteLog(t *testing.T) { t.Fatal(err) } - wal, idx, err := newWalWriter(f.Name()) + wal, err := newWalWriter(f.Name(), 0) if err != nil { t.Fatal(err) } - if got, want := idx, uint64(0); got != want { - t.Fatalf("expected index %d, got %d", want, got) - } reader, err := newWalReader(f.Name()) if err != nil { From 6002e55623266c5d92179144602b95f6b8d78d15 Mon Sep 17 00:00:00 2001 From: Martin Hutchinson Date: Thu, 31 Jul 2025 15:09:38 +0000 Subject: [PATCH 2/3] Updated milestones in README --- vindex/README.md | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/vindex/README.md b/vindex/README.md index 7a06989..cdaf7a4 100644 --- a/vindex/README.md +++ b/vindex/README.md @@ -221,14 +221,16 @@ go run ./vindex/cmd/client --base_url http://localhost:8088/vindex/ --lookup=foo | 1 | Public code base and documentation for prototype | ✅ | | 2 | Implementation of in-memory Merkle Radix Tree | ✅ | | 3 | Incremental update | ✅ | -| 4 | Example written for mapping SumDB | ✅ | -| 5 | Proofs served on Lookup | ❌ | -| 6 | Output log | ❌ | +| 4 | Verify that mapped data matches Input Log Checkpoint | ✅ | +| 5 | Output log | ❌ | +| 6 | Proofs served on Lookup | ❌ | | 7 | Storage backed verifiable-map | ❌ | -| 8 | Example written for mapping CT | ⚠️ | -| 9 | MapFn defined in WASM | ❌ | -| 10 | Proper repository for this code to live long-term | ❌ | -| 11 | Support reading directly from Input Log instead of Clone | ❌ | +| 8 | MapFn defined in WASM | ❌ | +| 9 | Proper repository for this code to live long-term | ❌ | +| 10 | Support reading directly from Input Log instead of Clone | ✅ | +| 11 | Example written for mapping SumDB | ✅ | +| 12 | Example written for hosting a log and VIndex together | ✅ | +| 13 | Example written for mapping CT | ⚠️ | | N | Production ready | ❌ | From e7aeb40b313438024dc1f9375dc5b908b72c6d13 Mon Sep 17 00:00:00 2001 From: Martin Hutchinson Date: Thu, 31 Jul 2025 15:54:11 +0000 Subject: [PATCH 3/3] Review comments --- vindex/cmd/logandmap/main.go | 4 ++-- vindex/map.go | 34 +++++++++++++++++----------------- vindex/map_test.go | 4 ++-- vindex/wal.go | 17 +++++++++-------- vindex/wal_test.go | 4 ++-- 5 files changed, 32 insertions(+), 31 deletions(-) diff --git a/vindex/cmd/logandmap/main.go b/vindex/cmd/logandmap/main.go index 6c41c12..331e9a6 100644 --- a/vindex/cmd/logandmap/main.go +++ b/vindex/cmd/logandmap/main.go @@ -281,7 +281,7 @@ func getKeyFile(path string) (string, error) { } func mapFnFromFlags() vindex.MapFn { - mapFn := func(data []byte) [][32]byte { + mapFn := func(data []byte) [][sha256.Size]byte { var entry LogEntry if err := json.Unmarshal(data, &entry); err != nil { panic(fmt.Errorf("failed to unmarshal entry: %v", err)) @@ -291,7 +291,7 @@ func mapFnFromFlags() vindex.MapFn { // This could be changed to return something more complex, e.g. include // a static prefix of "module=", which would allow the same map to host // multiple queries in parallel. - return [][32]byte{sha256.Sum256([]byte(entry.Module))} + return [][sha256.Size]byte{sha256.Sum256([]byte(entry.Module))} } return mapFn } diff --git a/vindex/map.go b/vindex/map.go index 72ed9c7..4f7a02b 100644 --- a/vindex/map.go +++ b/vindex/map.go @@ -42,9 +42,9 @@ import ( ) const ( - db_latestCheckpointKey = "latestCheckpoint" - db_compactRangeKey = "compactRange" - db_compactRangeSizeKey = "compactRangeSize" + dbLatestCheckpointKey = "latestCheckpoint" + dbCompactRangeKey = "compactRange" + dbCompactRangeSizeKey = "compactRangeSize" ) // MapFn takes the raw leaf data from a log entry and outputs the SHA256 hashes @@ -55,7 +55,7 @@ const ( // makes sense to the particular ecosystem. This might mean outputting any invalid leaves // at a known locations (e.g. all 0s), or not outputting any entry. Any panics will cause // the mapping process to terminate. -type MapFn func([]byte) [][32]byte +type MapFn func([]byte) [][sha256.Size]byte // InputLog represents a connection to the input log from which map data will be built. // This can be a local or remote data source. @@ -103,14 +103,14 @@ func NewVerifiableIndex(ctx context.Context, inputLog InputLog, inputLogParseFn snap := db.NewSnapshot() defer logClose(snap) - if sizeBs, sizeCloser, err := snap.Get([]byte(db_compactRangeSizeKey)); err != nil { + if sizeBs, sizeCloser, err := snap.Get([]byte(dbCompactRangeSizeKey)); err != nil { if err != pebble.ErrNotFound { return nil, fmt.Errorf("pebble.Get(): %v", err) } size = 0 cr = crf.NewEmptyRange(0) } else { - crBs, crCloser, err := snap.Get([]byte(db_compactRangeKey)) + crBs, crCloser, err := snap.Get([]byte(dbCompactRangeKey)) if err != nil { return nil, fmt.Errorf("pebble.Get(): %v", err) } @@ -155,7 +155,7 @@ func NewVerifiableIndex(ctx context.Context, inputLog InputLog, inputLogParseFn walReader: reader, db: db, vindex: *mpt.NewTree(sha256.Sum256, vtreeStorage), - data: map[[32]byte][]uint64{}, + data: map[[sha256.Size]byte][]uint64{}, } if err := b.buildMap(ctx); err != nil { return nil, fmt.Errorf("failed to build map: %v", err) @@ -214,7 +214,7 @@ func (m *inputLogMapper) syncFromInputLog(ctx context.Context) error { // Apply the MapFn in as safe a way as possible. This involves trapping any panics // and failing gracefully. - var hashes [][32]byte + var hashes [][sha256.Size]byte var mapErr error func() { defer func() { @@ -264,7 +264,7 @@ func (m *inputLogMapper) syncFromInputLog(ctx context.Context) error { if !bytes.Equal(hash, cp.Hash) { return fmt.Errorf("calculated hash for tree size %d is %x, but checkpoint commits to %x", cp.Size, hash, cp.Hash) } - if err := m.db.Set([]byte(db_latestCheckpointKey), rawCp, pebble.Sync); err != nil { + if err := m.db.Set([]byte(dbLatestCheckpointKey), rawCp, pebble.Sync); err != nil { return fmt.Errorf("failed to update state: %v", err) } @@ -273,15 +273,15 @@ func (m *inputLogMapper) syncFromInputLog(ctx context.Context) error { } func (m *inputLogMapper) storeState() error { - flatSlice := make([]byte, len(m.r.Hashes())*32) + flatSlice := make([]byte, len(m.r.Hashes())*sha256.Size) for i, arr := range m.r.Hashes() { - copy(flatSlice[i*32:], arr[:]) + copy(flatSlice[i*sha256.Size:], arr[:]) } b := m.db.NewBatch() - if err := b.Set([]byte(db_compactRangeKey), flatSlice, pebble.Sync); err != nil { + if err := b.Set([]byte(dbCompactRangeKey), flatSlice, pebble.Sync); err != nil { return fmt.Errorf("failed to update state: %v", err) } - if err := b.Set([]byte(db_compactRangeSizeKey), binary.BigEndian.AppendUint64(nil, m.r.End()), pebble.Sync); err != nil { + if err := b.Set([]byte(dbCompactRangeSizeKey), binary.BigEndian.AppendUint64(nil, m.r.End()), pebble.Sync); err != nil { return fmt.Errorf("failed to update state: %v", err) } if err := m.db.Apply(b, pebble.Sync); err != nil { @@ -299,7 +299,7 @@ type VerifiableIndex struct { indexMu sync.RWMutex // covers vindex and data vindex mpt.Tree - data map[[32]byte][]uint64 + data map[[sha256.Size]byte][]uint64 // servingSize is the size of the input log we are serving for. // This a temporary workaround not having an output log, which we will eventually read to get @@ -351,9 +351,9 @@ func (b *VerifiableIndex) Update(ctx context.Context) error { // built up the provided size. func (b *VerifiableIndex) buildMap(ctx context.Context) error { startWal := time.Now() - updatedKeys := make(map[[32]byte]bool) // Allows us to efficiently update vindex after first init + updatedKeys := make(map[[sha256.Size]byte]bool) // Allows us to efficiently update vindex after first init - cpRaw, closer, err := b.db.Get([]byte(db_latestCheckpointKey)) + cpRaw, closer, err := b.db.Get([]byte(dbLatestCheckpointKey)) if err != nil { if err == pebble.ErrNotFound { // If the key isn't there then nothing to do. @@ -431,7 +431,7 @@ func (b *VerifiableIndex) buildMap(ctx context.Context) error { } // Finally, we update the vindex - if err := b.vindex.Insert(h, [32]byte(sum.Sum(nil))); err != nil { + if err := b.vindex.Insert(h, [sha256.Size]byte(sum.Sum(nil))); err != nil { return fmt.Errorf("Insert(): %s", err) } } diff --git a/vindex/map_test.go b/vindex/map_test.go index 1c6074d..2ce0a10 100644 --- a/vindex/map_test.go +++ b/vindex/map_test.go @@ -57,12 +57,12 @@ func TestVerifiableIndex(t *testing.T) { cp, _, _, err := log.ParseCheckpoint(cpRaw, v.Name(), v) return cp, err } - mapFn := func(leaf []byte) [][32]byte { + mapFn := func(leaf []byte) [][sha256.Size]byte { key, _, found := bytes.Cut(leaf, []byte(":")) if !found { panic("colon not found") } - return [][32]byte{sha256.Sum256(key)} + return [][sha256.Size]byte{sha256.Sum256(key)} } f, err := os.CreateTemp("", "vindexTestDir") if err != nil { diff --git a/vindex/wal.go b/vindex/wal.go index d4b505e..2630598 100644 --- a/vindex/wal.go +++ b/vindex/wal.go @@ -16,6 +16,7 @@ package vindex import ( "bufio" + "crypto/sha256" "encoding/hex" "errors" "fmt" @@ -162,7 +163,7 @@ func (l *walWriter) flush() error { return l.f.Sync() } -func (l *walWriter) append(idx uint64, hashes [][32]byte) error { +func (l *walWriter) append(idx uint64, hashes [][sha256.Size]byte) error { e, err := marshalWalEntry(idx, hashes) if err != nil { return fmt.Errorf("failed to marshal entry: %v", err) @@ -192,7 +193,7 @@ type walReader struct { // TODO(mhutchinson): change this as it's inconvenient with EOF handling, // which should be common when reader hits the end of the file but more is // to be written. -func (r *walReader) next() (uint64, [][32]byte, error) { +func (r *walReader) next() (uint64, [][sha256.Size]byte, error) { line, err := r.r.ReadString('\n') if err != nil { if err == io.EOF { @@ -213,23 +214,23 @@ func (r *walReader) close() error { // unmarshalWalEntry parses a line from the WAL. // This is the reverse of marshalWalEntry. -func unmarshalWalEntry(e string) (uint64, [][32]byte, error) { +func unmarshalWalEntry(e string) (uint64, [][sha256.Size]byte, error) { tokens := strings.Split(e, " ") idx, err := strconv.ParseUint(tokens[0], 10, 64) if err != nil { return 0, nil, fmt.Errorf("failed to parse idx from %q", e) } - hashes := make([][32]byte, 0, len(tokens)-1) + hashes := make([][sha256.Size]byte, 0, len(tokens)-1) for i, h := range tokens[1:] { parsed, err := hex.DecodeString(h) if err != nil { return 0, nil, fmt.Errorf("failed to parse hex token %d from %q", i, e) } - if got, want := len(parsed), 32; got != want { - return 0, nil, fmt.Errorf("expected 32 byte hash but got %d bytes at idx %d", got, i) + if got, want := len(parsed), sha256.Size; got != want { + return 0, nil, fmt.Errorf("expected %d byte hash but got %d bytes at idx %d", want, got, i) } - hashes = append(hashes, [32]byte(parsed)) + hashes = append(hashes, [sha256.Size]byte(parsed)) } return idx, hashes, nil @@ -237,7 +238,7 @@ func unmarshalWalEntry(e string) (uint64, [][32]byte, error) { // unmarshalWalEntry converts an index and the hashes it affects into a line for the WAL. // This is the reverse of unmarshalWalEntry. -func marshalWalEntry(idx uint64, hashes [][32]byte) (string, error) { +func marshalWalEntry(idx uint64, hashes [][sha256.Size]byte) (string, error) { sb := strings.Builder{} if _, err := sb.WriteString(strconv.FormatUint(idx, 10)); err != nil { return "", err diff --git a/vindex/wal_test.go b/vindex/wal_test.go index 7d050e0..da87d35 100644 --- a/vindex/wal_test.go +++ b/vindex/wal_test.go @@ -168,7 +168,7 @@ func TestWriteAheadLog_roundtrip(t *testing.T) { for i := range tC.entriesToWrite { hash := sha256.Sum256([]byte{byte(i)}) - if err := wal.append(uint64(i), [][32]byte{hash}); err != nil { + if err := wal.append(uint64(i), [][sha256.Size]byte{hash}); err != nil { t.Error(err) } } @@ -233,7 +233,7 @@ func TestWriteAndWriteLog(t *testing.T) { eg.Go(func() error { for i := range count { hash := sha256.Sum256([]byte{byte(i)}) - if err := wal.append(uint64(i), [][32]byte{hash}); err != nil { + if err := wal.append(uint64(i), [][sha256.Size]byte{hash}); err != nil { return err } }