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/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 | ❌ | diff --git a/vindex/cmd/logandmap/main.go b/vindex/cmd/logandmap/main.go index 640bb13..331e9a6 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 { @@ -275,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)) @@ -285,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 b95b4f1..4f7a02b 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 ( + dbLatestCheckpointKey = "latestCheckpoint" + dbCompactRangeKey = "compactRange" + dbCompactRangeSizeKey = "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). @@ -42,7 +55,7 @@ import ( // 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. @@ -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(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(dbCompactRangeKey)) + 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{}, + data: map[[sha256.Size]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,18 +187,34 @@ 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. - var hashes [][32]byte + var hashes [][sha256.Size]byte var mapErr error func() { defer func() { @@ -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(dbLatestCheckpointKey), 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())*sha256.Size) + for i, arr := range m.r.Hashes() { + copy(flatSlice[i*sha256.Size:], arr[:]) + } + b := m.db.NewBatch() + 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(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 { + return fmt.Errorf("failed to update state: %v", err) + } return nil } @@ -187,10 +295,11 @@ 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 - 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 @@ -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 + updatedKeys := make(map[[sha256.Size]byte]bool) // Allows us to efficiently update vindex after first init + + 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. + 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 < toSize; { + for i := b.servingSize; i < size; { select { case <-ctx.Done(): return ctx.Err() @@ -313,14 +431,40 @@ func (b *VerifiableIndex) buildMap(ctx context.Context, toSize uint64) 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) } } 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..2ce0a10 100644 --- a/vindex/map_test.go +++ b/vindex/map_test.go @@ -57,20 +57,26 @@ 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("", "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..2630598 100644 --- a/vindex/wal.go +++ b/vindex/wal.go @@ -16,6 +16,7 @@ package vindex import ( "bufio" + "crypto/sha256" "encoding/hex" "errors" "fmt" @@ -27,46 +28,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 +75,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 +103,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,36 +114,56 @@ 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) append(idx uint64, hashes [][32]byte) error { +func (l *walWriter) flush() error { + return l.f.Sync() +} + +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) @@ -174,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 { @@ -195,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 @@ -219,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 17d60b1..da87d35 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), [][sha256.Size]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 { @@ -204,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 } }