Add lease tracking to bonanza_storage_shard

Our disk based storage backend (which is still a stub) currently needs
to deal with storing objects and tracking leases. However in our case
it's sufficient to just have a decorator that does the lease tracking
for us.

Solve this by adding a new FlatStore interface, which is essentially a
simplified version of Store that uses a new FlatReference type.
FlatReference is similar to an REv2 digest, only storing a hash and
size. We then add an adapter on top of FlatStore that does the lease
tracking.

The next step will be to add a proper FlatStore implementation that
writes objects to disk. This can be a lot simpler than LocalBlobAccess,
because we don't need to support FindMissingBlobs() and only need to
store objects up to 2 MB in size.

Author: EdSchouten

cmd/bonanza_storage_shard/BUILD.bazel

@@ -7,16 +7,19 @@ go_library(
     importpath = "github.com/buildbarn/bonanza/cmd/bonanza_storage_shard",
     visibility = ["//visibility:private"],
     deps = [
+        "//pkg/ds/lossymap",
         "//pkg/proto/configuration/bonanza_storage_shard",
         "//pkg/proto/storage/object",
         "//pkg/proto/storage/tag",
         "//pkg/storage/object",
+        "//pkg/storage/object/flatbacked",
         "//pkg/storage/object/leasemarshaling",
         "//pkg/storage/object/local",
         "//pkg/storage/object/namespacemapping",
         "//pkg/storage/tag",
         "//pkg/storage/tag/leasemarshaling",
         "//pkg/storage/tag/local",
+        "@com_github_buildbarn_bb_storage//pkg/clock",
         "@com_github_buildbarn_bb_storage//pkg/global",
         "@com_github_buildbarn_bb_storage//pkg/grpc",
         "@com_github_buildbarn_bb_storage//pkg/program",

cmd/bonanza_storage_shard/main.go

@@ -2,16 +2,20 @@ package main
 
 import (
 	"context"
+	"math/rand/v2"
 	"os"
 
+	"github.com/buildbarn/bb-storage/pkg/clock"
 	"github.com/buildbarn/bb-storage/pkg/global"
 	bb_grpc "github.com/buildbarn/bb-storage/pkg/grpc"
 	"github.com/buildbarn/bb-storage/pkg/program"
 	"github.com/buildbarn/bb-storage/pkg/util"
+	"github.com/buildbarn/bonanza/pkg/ds/lossymap"
 	"github.com/buildbarn/bonanza/pkg/proto/configuration/bonanza_storage_shard"
 	object_pb "github.com/buildbarn/bonanza/pkg/proto/storage/object"
 	tag_pb "github.com/buildbarn/bonanza/pkg/proto/storage/tag"
 	"github.com/buildbarn/bonanza/pkg/storage/object"
+	object_flatbacked "github.com/buildbarn/bonanza/pkg/storage/object/flatbacked"
 	object_leasemarshaling "github.com/buildbarn/bonanza/pkg/storage/object/leasemarshaling"
 	object_local "github.com/buildbarn/bonanza/pkg/storage/object/local"
 	object_namespacemapping "github.com/buildbarn/bonanza/pkg/storage/object/namespacemapping"
@@ -38,9 +42,56 @@ func main() {
 			return util.StatusWrap(err, "Failed to apply global configuration options")
 		}
 
-		objectStore := object_local.NewStore()
+		// Construct a flat storage backend for objects. Put an
+		// in-memory store for leases in front of it, so that
+		// UploadDag() can reliably enforce that clients upload
+		// complete DAGs.
+		leaseCompletenessDuration := configuration.LeasesMapLeaseCompletenessDuration
+		if err := leaseCompletenessDuration.CheckValid(); err != nil {
+			return util.StatusWrap(err, "Invalid leases map lease completeness duration")
+		}
+		leaseComparator := func(a, b *object_flatbacked.Lease) int {
+			if *a < *b {
+				return -1
+			}
+			if *a > *b {
+				return 1
+			}
+			return 0
+		}
+		hashInitialization := rand.Uint64()
+		objectStore := object_flatbacked.NewStore(
+			object_local.NewStore(),
+			clock.SystemClock,
+			leaseCompletenessDuration.AsDuration(),
+			lossymap.NewHashMap(
+				lossymap.NewSimpleRecordArray[object.LocalReference](
+					int(configuration.LeasesMapRecordsCount),
+					leaseComparator,
+				),
+				/* recordKeyHasher = */ func(k *lossymap.RecordKey[object.LocalReference]) uint64 {
+					h := hashInitialization
+					for _, c := range k.Key.GetRawReference() {
+						h ^= uint64(c)
+						h *= 1099511628211
+					}
+					attempt := k.Attempt
+					for i := 0; i < 4; i++ {
+						h ^= uint64(attempt & 0xff)
+						h *= 1099511628211
+						attempt >>= 8
+					}
+					return h
+				},
+				configuration.LeasesMapRecordsCount,
+				leaseComparator,
+				configuration.LeasesMapMaximumGetAttempts,
+				int(configuration.LeasesMapMaximumPutAttempts),
+				"LeasesMap",
+			),
+		)
 		tagStore := tag_local.NewStore()
-		leaseMarshaler := object_local.LeaseMarshaler
+		leaseMarshaler := object_flatbacked.LeaseMarshaler
 
 		if err := bb_grpc.NewServersFromConfigurationAndServe(
 			configuration.GrpcServers,

deployments/demo/bonanza_storage_shard.jsonnet

@@ -7,4 +7,9 @@ local shard = std.extVar('SHARD');
     listenPaths: ['%s/bonanza_storage_shard_%s%s.sock' % [statePath, replica, shard]],
     authenticationPolicy: { allow: {} },
   }],
+
+  leasesMapRecordsCount: 1e6,
+  leasesMapLeaseCompletenessDuration: '120s',
+  leasesMapMaximumGetAttempts: 16,
+  leasesMapMaximumPutAttempts: 64,
 }

pkg/proto/configuration/bonanza_storage_shard/BUILD.bazel

@@ -9,6 +9,7 @@ proto_library(
     deps = [
         "@com_github_buildbarn_bb_storage//pkg/proto/configuration/global:global_proto",
         "@com_github_buildbarn_bb_storage//pkg/proto/configuration/grpc:grpc_proto",
+        "@protobuf//:duration_proto",
     ],
 )
 

pkg/proto/configuration/bonanza_storage_shard/bonanza_storage_shard.pb.go

@@ -2,6 +2,7 @@ syntax = "proto3";
 
 package bonanza.configuration.bonanza_storage_shard;
 
+import "google/protobuf/duration.proto";
 import "pkg/proto/configuration/global/global.proto";
 import "pkg/proto/configuration/grpc/grpc.proto";
 
@@ -13,4 +14,74 @@ message ApplicationConfiguration {
 
   // gRPC servers to spawn to listen for requests from clients.
   repeated buildbarn.configuration.grpc.ServerConfiguration grpc_servers = 2;
+
+  // The size of the leases map.
+  //
+  // The leases map is an in-memory data structure that stores leases of
+  // all children of objects stored in the current shard. These leases
+  // track the completeness of graphs of objects, which is necessary for
+  // UploadDag() operations against the storage frontend to run
+  // efficiently.
+  //
+  // Each record of the leases map is less than 64 bytes in size, and
+  // records are stored contiguously. This means that it is generally
+  // safe to set this option to millions or more.
+  //
+  // The leases map needs to be large enough to not lose any records
+  // that haven't expired yet. The following Prometheus queries may be
+  // used to determine insertions into the leases map caused other
+  // records to be displaced prematurely:
+  //
+  // bonanza_lossymap_hash_map_put_iterations_count{name="LeasesMap",outcome="TooManyAttempts"}
+  // bonanza_lossymap_hash_map_put_too_many_iterations_total{name="LeasesMap"}
+  uint64 leases_map_records_count = 3;
+
+  // The maximum age a lease may have for a parent object using the
+  // lease to be considered complete.
+  //
+  // If a parent object references one or more children for which the
+  // lease's age exceeds twice this value, the storage frontend is
+  // instructed to renew the leases whose age exceeds this value.
+  //
+  // More concretely speaking, consider the case in which this option is
+  // set to the recommended value of 2 minutes. In this case a parent
+  // object is considered to be complete if the leases of all of its
+  // children are less than 4 minutes old. If one or more leases are
+  // more than 4 minutes old (or missing), the storage frontend has to
+  // renew all the leases that are at least 2 minutes old (or
+  // missing) to make the parent object complete again.
+  //
+  // What this means in practice is that for UploadDag() to work
+  // reliably (i.e., not reporting graphs as being present while in
+  // reality they are not), a storage replica may only be restarted if
+  // the other replica is healthy for at least
+  // 'height*leases_map_lease_completeness_duration' time, where
+  // 'height' refers to the height of the root node of the DAG that is
+  // being uploaded.
+  //
+  // Recommended value: 120s
+  google.protobuf.Duration leases_map_lease_completeness_duration = 4;
+
+  // The number of records a Get() call on the leases map may attempt
+  // to access.
+  //
+  // The lower the utilization rate of the leases map, the lower this
+  // value may be set. For example, if the size of the leases map is set
+  // in such a way that it is only utilized by 10% (factor 0.1), setting
+  // this field to 16 means there is only a 0.1^16 chance that inserting
+  // a record prematurely displaces another valid lease.
+  //
+  // Recommended value: 16
+  uint32 leases_map_maximum_get_attempts = 5;
+
+  // The number of mutations that a Put() on the leases map may perform.
+  //
+  // Because the leases map uses a scheme similar to Robin Hood hashing,
+  // insertions may cause other records to be displaced. Those records
+  // may then cause even more records to be displaced. Because of that,
+  // it is recommended to set this field to a small multiple of the
+  // maximum Get() attempts.
+  //
+  // Recommended value: 64
+  int64 leases_map_maximum_put_attempts = 6;
 }

pkg/proto/configuration/bonanza_storage_shard/bonanza_storage_shard.proto

@@ -8,6 +8,7 @@ go_library(
         "contents.go",
         "downloader.go",
         "downloader_server.go",
+        "flat_reference.go",
         "global_reference.go",
         "instance_name.go",
         "limit.go",

pkg/storage/object/BUILD.bazel

@@ -15,6 +15,7 @@ type BasicReference interface {
 	GetReferenceFormat() ReferenceFormat
 	GetRawReference() []byte
 
-	// Conversion to a LocalReference with the same value.
+	// Conversion to other reference types with the same value.
 	GetLocalReference() LocalReference
+	Flatten() FlatReference
 }

pkg/storage/object/basic_reference.go

@@ -107,15 +107,15 @@ func (c *Contents) cloneWithReference(r LocalReference) *Contents {
 	}
 }
 
-// Flatten the reference of the object, so that its height and degree
-// are both zero.
+// FlattenContents flattens the reference of the object, so that its
+// height and degree are both zero.
 //
 // This is used by the read caching backend, where we want to cache
 // individual objects as opposed to graphs. When writing objects to the
 // local cache, we set the height and degree to zero, so that there is
 // no need to track any leases.
-func (c *Contents) Flatten() *Contents {
-	return c.cloneWithReference(c.LocalReference.Flatten())
+func (c *Contents) FlattenContents() *Contents {
+	return c.cloneWithReference(c.LocalReference.Flatten().GetLocalReference())
 }
 
 func (c *Contents) validateOutgoingReferences() error {

pkg/storage/object/contents.go

@@ -0,0 +1,34 @@
+package object
+
+import (
+	"encoding/binary"
+	"encoding/hex"
+)
+
+// FlatReference is a simplified version of LocalReference that assumes
+// that the height and degree of an object is zero. This is sufficient
+// for the read caching backend, which is oblivious of references
+// between objects.
+type FlatReference struct {
+	rawReference [35]byte
+}
+
+// MustNewSHA256V1LocalReference creates a flat reference that uses
+// reference format SHA256_V1. This function can be used as part of
+// tests.
+func MustNewSHA256V1FlatReference(hash string, sizeBytes uint32) (flatReference FlatReference) {
+	var rawReference [36]byte
+	if n, err := hex.Decode(rawReference[:], []byte(hash)); err != nil {
+		panic(err)
+	} else if n != 32 {
+		panic("Wrong hash length")
+	}
+	binary.LittleEndian.PutUint32(rawReference[32:], sizeBytes)
+	copy(flatReference.rawReference[:], rawReference[:])
+	return
+}
+
+func (r FlatReference) GetLocalReference() (localReference LocalReference) {
+	copy(localReference.rawReference[:], r.rawReference[:])
+	return
+}