Skip to content

Replace CAS with CS and CLS #347

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
meroton-benjamin wants to merge 6 commits into
base: main
Choose a base branch
from
Draft

Replace CAS with CS and CLS #347

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
5eb0772
Add a generic message reader interface
meroton-benjamin Aug 22, 2026
9a21c6a
Add support for SplitBlob and SpliceBlob methods
meroton-benjamin May 29, 2026
10b73fd
Replace CAS with CS and CLS
meroton-benjamin Jun 9, 2026
2f4d4b7
Move ContentAddressableStorage to pkg/cas
meroton-benjamin Aug 24, 2026
0e85673
Remove debug print
meroton-benjamin Aug 26, 2026
004717c
Replace testutils
meroton-benjamin Aug 26, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions MODULE.bazel
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@ use_repo(
"com_github_aws_aws_sdk_go_v2_service_sts",
"com_github_bazelbuild_buildtools",
"com_github_bazelbuild_remote_apis",
"com_github_buildbarn_go_cdc",
"com_github_buildbarn_go_sha256tree",
"com_github_fxtlabs_primes",
"com_github_go_jose_go_jose_v3",
Expand Down
4 changes: 2 additions & 2 deletions cmd/bb_copy/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ func main() {

grpcClientFactory := grpc.NewBaseClientFactory(grpc.BaseClientDialer, nil, nil, nil)

blobAccessCreator := blobstore_configuration.NewCASBlobAccessCreator(
blobAccessCreator := blobstore_configuration.NewCSBlobAccessCreator(

@EdSchouten EdSchouten Aug 18, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Keep in mind that bb_copy can, for example, be used to make backups of actions. This means that for bb_copy it's not sufficient to only use a Chunk Store. It should also preserve/expand chunk lists, etc.

grpcClientFactory,
int(configuration.MaximumMessageSizeBytes),
bb_zstd.NewPoolFromConfiguration(nil),
Expand All @@ -68,7 +68,7 @@ func main() {
configuration.Replicator,
source.BlobAccess,
sink,
blobstore_configuration.NewCASBlobReplicatorCreator(grpcClientFactory),
blobstore_configuration.NewCSBlobReplicatorCreator(grpcClientFactory),
)
if err != nil {
return util.StatusWrap(err, "Failed to create replicator")
Expand Down
4 changes: 2 additions & 2 deletions cmd/bb_replicator/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ func main() {
return util.StatusWrap(err, "Failed to apply global configuration options")

@EdSchouten EdSchouten Aug 18, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Now that we only deal with chunks and no longer do any streaming, do you think we still need bb_replicator? Maybe it's fine to just throw it out altogether, and let bb_frontend do the fixups directly?

}

blobAccessCreator := blobstore_configuration.NewCASBlobAccessCreator(
blobAccessCreator := blobstore_configuration.NewCSBlobAccessCreator(
grpcClientFactory,
int(configuration.MaximumMessageSizeBytes),
bb_zstd.NewPoolFromConfiguration(nil),
Expand All @@ -59,7 +59,7 @@ func main() {
configuration.Replicator,
source.BlobAccess,
sink,
blobstore_configuration.NewCASBlobReplicatorCreator(grpcClientFactory),
blobstore_configuration.NewCSBlobReplicatorCreator(grpcClientFactory),
)
if err != nil {
return util.StatusWrap(err, "Failed to create replicator")
Expand Down
2 changes: 2 additions & 0 deletions cmd/bb_storage/BUILD.bazel
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,10 +10,12 @@ go_library(
"//pkg/auth",
"//pkg/auth/configuration",
"//pkg/blobstore",
"//pkg/blobstore/cdc",
"//pkg/blobstore/configuration",
"//pkg/blobstore/grpcservers",
"//pkg/builder",
"//pkg/capabilities",
"//pkg/cas",
"//pkg/global",
"//pkg/grpc",
"//pkg/program",
Expand Down
90 changes: 64 additions & 26 deletions cmd/bb_storage/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,10 +9,12 @@ import (
"github.com/buildbarn/bb-storage/pkg/auth"
auth_configuration "github.com/buildbarn/bb-storage/pkg/auth/configuration"
"github.com/buildbarn/bb-storage/pkg/blobstore"
"github.com/buildbarn/bb-storage/pkg/blobstore/cdc"
blobstore_configuration "github.com/buildbarn/bb-storage/pkg/blobstore/configuration"
"github.com/buildbarn/bb-storage/pkg/blobstore/grpcservers"
"github.com/buildbarn/bb-storage/pkg/builder"
"github.com/buildbarn/bb-storage/pkg/capabilities"
"github.com/buildbarn/bb-storage/pkg/cas"
"github.com/buildbarn/bb-storage/pkg/global"
bb_grpc "github.com/buildbarn/bb-storage/pkg/grpc"
"github.com/buildbarn/bb-storage/pkg/program"
Expand Down Expand Up @@ -54,34 +56,58 @@ func main() {
var cacheCapabilitiesAuthorizers []auth.Authorizer

// Content Addressable Storage (CAS).
var contentAddressableStorageInfo *blobstore_configuration.BlobAccessInfo
var contentAddressableStorage blobstore.BlobAccess
if configuration.ContentAddressableStorage != nil {
info, authorizedBackend, allAuthorizers, err := newScannableBlobAccess(
var contentAddressableStorage cas.ContentAddressableStorage
var authorizedContentAddressableStorage cas.ContentAddressableStorage
if configuration.ContentAddressableStorageServer != nil {
var chunkStorage, chunkListStorage blobstore.BlobAccess
var cdcParametersFetcher cdc.ParametersFetcher
contentAddressableStorage, chunkStorage, chunkListStorage, _, cdcParametersFetcher, err = blobstore_configuration.NewCASFromConfiguration(
dependenciesGroup,
configuration.ContentAddressableStorage,
blobstore_configuration.NewCASBlobAccessCreator(
grpcClientFactory,
int(configuration.MaximumMessageSizeBytes),
zstdPool,
),
configuration.ContentAddressableStorageServer.ContentAddressableStorage,
grpcClientFactory,
int(configuration.MaximumMessageSizeBytes),
zstdPool,
)
if err != nil {
return util.StatusWrap(err, "Failed to create Content Addressable Storage")
}

// Create authorizers.
getAuthorizer, err := auth_configuration.DefaultAuthorizerFactory.NewAuthorizerFromConfiguration(configuration.ContentAddressableStorageServer.GetAuthorizer, dependenciesGroup, grpcClientFactory)
if err != nil {
return util.StatusWrap(err, "Failed to create Get() authorizer for Content Addressable Storage")
}
putAuthorizer, err := auth_configuration.DefaultAuthorizerFactory.NewAuthorizerFromConfiguration(configuration.ContentAddressableStorageServer.PutAuthorizer, dependenciesGroup, grpcClientFactory)
if err != nil {
return util.StatusWrap(err, "Failed to create Put() authorizer for Content Addressable Storage")
}
findMissingAuthorizer, err := auth_configuration.DefaultAuthorizerFactory.NewAuthorizerFromConfiguration(configuration.ContentAddressableStorageServer.FindMissingAuthorizer, dependenciesGroup, grpcClientFactory)
if err != nil {
return util.StatusWrap(err, "Failed to create FindMissing() authorizer for Content Addressable Storage")
}

// Create authorized versions of the backends.
authorizedChunkStorage := blobstore.NewAuthorizingBlobAccess(chunkStorage, getAuthorizer, putAuthorizer, findMissingAuthorizer)
authorizedChunkListStorage := blobstore.NewAuthorizingBlobAccess(chunkListStorage, getAuthorizer, putAuthorizer, findMissingAuthorizer)
authorizedChunkListFetcher := blobstore.NewBlobAccessChunkListFetcher(authorizedChunkListStorage, int(configuration.MaximumMessageSizeBytes))
authorizedContentAddressableStorage = cas.NewContentAddressableStorage(
authorizedChunkStorage,
authorizedChunkListStorage,
authorizedChunkListFetcher,
cdcParametersFetcher,
contentAddressableStorage.GetDigestKeyFormat(),
)
// Create the Chunk Storage (CS).
cacheCapabilitiesProviders = append(
cacheCapabilitiesProviders,
info.BlobAccess,
chunkStorage,
capabilities.NewStaticProvider(&remoteexecution.ServerCapabilities{
CacheCapabilities: &remoteexecution.CacheCapabilities{
SupportedCompressors: configuration.SupportedCompressors,
},
}),
)
cacheCapabilitiesAuthorizers = append(cacheCapabilitiesAuthorizers, allAuthorizers...)
contentAddressableStorageInfo = &info
contentAddressableStorage = authorizedBackend
cacheCapabilitiesAuthorizers = append(cacheCapabilitiesAuthorizers, getAuthorizer, putAuthorizer, findMissingAuthorizer)
}

// Action Cache (AC).
Expand All @@ -91,7 +117,7 @@ func main() {
dependenciesGroup,
configuration.ActionCache,
blobstore_configuration.NewACBlobAccessCreator(
contentAddressableStorageInfo,
contentAddressableStorage,
grpcClientFactory,
int(configuration.MaximumMessageSizeBytes),
),
Expand Down Expand Up @@ -192,19 +218,19 @@ func main() {
if err := bb_grpc.NewServersFromConfigurationAndServe(
configuration.GrpcServers,
func(s grpc.ServiceRegistrar) {
if contentAddressableStorage != nil {
if authorizedContentAddressableStorage != nil {
contentAddressableStorageServer := grpcservers.NewContentAddressableStorageServer(
authorizedContentAddressableStorage,
configuration.MaximumMessageSizeBytes,
)
remoteexecution.RegisterContentAddressableStorageServer(
s,
grpcservers.NewContentAddressableStorageServer(
contentAddressableStorage,
configuration.MaximumMessageSizeBytes,
),
contentAddressableStorageServer,
)
bytestream.RegisterByteStreamServer(
s,
grpcservers.NewByteStreamServer(
contentAddressableStorage,
1<<16,
authorizedContentAddressableStorage,
zstdPool,
),
)
Expand All @@ -214,7 +240,10 @@ func main() {
s,
grpcservers.NewActionCacheServer(
actionCache,
int(configuration.MaximumMessageSizeBytes),
blobstore.NewBlobAccessMessageReader[*remoteexecution.ActionResult](
actionCache,
int(configuration.MaximumMessageSizeBytes),
),
),
)
}
Expand All @@ -223,7 +252,10 @@ func main() {
s,
grpcservers.NewIndirectContentAddressableStorageServer(
indirectContentAddressableStorage,
int(configuration.MaximumMessageSizeBytes),
blobstore.NewBlobAccessMessageReader[*icas.Reference](
indirectContentAddressableStorage,
int(configuration.MaximumMessageSizeBytes),
),
),
)
}
Expand All @@ -232,7 +264,10 @@ func main() {
s,
grpcservers.NewInitialSizeClassCacheServer(
initialSizeClassCache,
int(configuration.MaximumMessageSizeBytes),
blobstore.NewBlobAccessMessageReader[*iscc.PreviousExecutionStats](
initialSizeClassCache,
int(configuration.MaximumMessageSizeBytes),
),
),
)
}
Expand All @@ -241,7 +276,10 @@ func main() {
s,
grpcservers.NewFileSystemAccessCacheServer(
fileSystemAccessCache,
int(configuration.MaximumMessageSizeBytes),
blobstore.NewBlobAccessMessageReader[*fsac.FileSystemAccessProfile](
fileSystemAccessCache,
int(configuration.MaximumMessageSizeBytes),
),
),
)
}
Expand Down
1 change: 1 addition & 0 deletions go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ require (
github.com/bazelbuild/buildtools v0.0.0-20260527135131-3b47c424ecf5
github.com/bazelbuild/remote-apis v0.0.0-20260331222004-becdd8f9ff81
github.com/bazelbuild/rules_go v0.62.0
github.com/buildbarn/go-cdc v0.0.9
github.com/buildbarn/go-sha256tree v0.0.0-20250310211320-0f70f20e855b
github.com/fxtlabs/primes v0.0.0-20150821004651-dad82d10a449
github.com/go-jose/go-jose/v3 v3.0.5
Expand Down
2 changes: 2 additions & 0 deletions go.sum
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,8 @@ github.com/bazelbuild/rules_go v0.62.0/go.mod h1:6YghDRf6l3FSiAwncK+Ww9jE1naoQzN
github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
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/buildbarn/go-cdc v0.0.9 h1:bWfgn92ed8Oo2zZKJdMAfB0APGz7Q8zvnqUn3hPuihM=
github.com/buildbarn/go-cdc v0.0.9/go.mod h1:KUMqSMvoRlby3uak9aKIvgz3KgNqwm2CMUoVX1EDr8k=
github.com/buildbarn/go-sha256tree v0.0.0-20250310211320-0f70f20e855b h1:IKUxixGBm9UxobU7c248z0BF0ojG19uoSLz8MFZM/KA=
github.com/buildbarn/go-sha256tree v0.0.0-20250310211320-0f70f20e855b/go.mod h1:e7g3/yWApcg+PpDqd4eQEEV8pexQmfCgK3frP+1Wuvk=
github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
Expand Down
41 changes: 41 additions & 0 deletions internal/mock/BUILD.bazel
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -152,6 +152,30 @@ gomock(
package = "mock",
)

gomock(
name = "cas",
out = "cas.go",
interfaces = [
"ContentAddressableStorage",
],
library = "//pkg/cas",
mockgen_model_library = "@org_uber_go_mock//mockgen/model",
mockgen_tool = "@org_uber_go_mock//mockgen",
package = "mock",
)

gomock(
name = "cdc",
out = "cdc.go",
interfaces = [
"ParametersFetcher",
],
library = "//pkg/blobstore/cdc",
mockgen_model_library = "@org_uber_go_mock//mockgen/model",
mockgen_tool = "@org_uber_go_mock//mockgen",
package = "mock",
)

gomock(
name = "clock",
out = "clock.go",
Expand Down Expand Up @@ -320,6 +344,16 @@ gomock(
package = "mock",
)

gomock(
name = "storage",
out = "storage.go",
interfaces = ["MessageReader"],
library = "//pkg/storage",
mockgen_tool = "@org_uber_go_mock//mockgen",
package = "mock",
source = "//pkg/storage:message_reader.go",
)

gomock(
name = "trace",
out = "trace.go",
Expand Down Expand Up @@ -367,6 +401,8 @@ go_library(
"buffer.go",
"builder.go",
"capabilities.go",
"cas.go",
"cdc.go",
"clock.go",
"cloud_aws.go",
"cloud_gcp.go",
Expand All @@ -380,6 +416,7 @@ go_library(
"prometheus.go",
"random.go",
"remoteexecution.go",
"storage.go",
"trace.go",
"trace_wrap.go",
"util.go",
Expand All @@ -391,10 +428,13 @@ go_library(
"//pkg/auth",
"//pkg/blobstore",
"//pkg/blobstore/buffer",
"//pkg/blobstore/cdc",
"//pkg/blobstore/chunklist",
"//pkg/blobstore/local",
"//pkg/blobstore/sharding",
"//pkg/blobstore/slicing",
"//pkg/builder",
"//pkg/cas",
"//pkg/clock",
"//pkg/cloud/gcp",
"//pkg/digest",
Expand All @@ -416,6 +456,7 @@ go_library(
"@org_golang_google_grpc//:grpc",
"@org_golang_google_grpc//metadata",
"@org_golang_google_protobuf//encoding/protowire",
"@org_golang_google_protobuf//proto",
"@org_uber_go_mock//gomock",
],
)
19 changes: 6 additions & 13 deletions pkg/blobstore/BUILD.bazel
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,10 @@ go_library(
"action_result_timestamp_injecting_blob_access.go",
"authorizing_blob_access.go",
"blob_access.go",
"blob_access_message_reader.go",
"cas_read_buffer_factory.go",
"chunk_list_fetcher.go",
"cls_read_buffer_factory.go",
"deadline_enforcing_blob_access.go",
"demultiplexing_blob_access.go",
"empty_blob_injecting_blob_access.go",
Expand All @@ -21,7 +24,6 @@ go_library(
"metrics_blob_access.go",
"read_buffer_factory.go",
"read_canarying_blob_access.go",
"reference_expanding_blob_access.go",
"validation_caching_read_buffer_factory.go",
"visit_topologically_sorted_tree.go",
"zip_reading_blob_access.go",
Expand All @@ -32,21 +34,19 @@ go_library(
deps = [
"//pkg/auth",
"//pkg/blobstore/buffer",
"//pkg/blobstore/chunklist",
"//pkg/blobstore/slicing",
"//pkg/capabilities",
"//pkg/clock",
"//pkg/cloud/aws",
"//pkg/cloud/gcp",
"//pkg/digest",
"//pkg/eviction",
"//pkg/proto/blobstore/chunklist",
"//pkg/proto/fsac",
"//pkg/proto/icas",
"//pkg/proto/iscc",
"//pkg/storage",
"//pkg/util",
"//pkg/zstd",
"@bazel_remote_apis//build/bazel/remote/execution/v2:remote_execution_go_proto",
"@com_github_aws_aws_sdk_go_v2//aws",
"@com_github_aws_aws_sdk_go_v2_service_s3//:s3",
"@com_github_prometheus_client_golang//prometheus",
"@org_golang_google_grpc//codes",
"@org_golang_google_grpc//status",
Expand All @@ -67,7 +67,6 @@ go_test(
"existence_caching_blob_access_test.go",
"hierarchical_instance_names_blob_access_test.go",
"read_canarying_blob_access_test.go",
"reference_expanding_blob_access_test.go",
"validation_caching_read_buffer_factory_test.go",
"visit_topologically_sorted_tree_test.go",
"zip_reading_blob_access_test.go",
Expand All @@ -79,16 +78,10 @@ go_test(
"//pkg/blobstore/buffer",
"//pkg/digest",
"//pkg/eviction",
"//pkg/proto/icas",
"//pkg/testutil",
"//pkg/util",
"//pkg/zstd",
"@bazel_remote_apis//build/bazel/remote/execution/v2:remote_execution_go_proto",
"@bazel_remote_apis//build/bazel/semver:semver_go_proto",
"@com_github_aws_aws_sdk_go_v2//aws",
"@com_github_aws_aws_sdk_go_v2_service_s3//:s3",
"@com_github_aws_aws_sdk_go_v2_service_s3//types",
"@com_github_klauspost_compress//zstd",
"@com_github_stretchr_testify//require",
"@org_golang_google_grpc//codes",
"@org_golang_google_grpc//status",
Expand Down
Loading