diff --git a/effects/cloud_read_miss_test.go b/effects/cloud_read_miss_test.go index 2851649..79e847c 100644 --- a/effects/cloud_read_miss_test.go +++ b/effects/cloud_read_miss_test.go @@ -82,8 +82,10 @@ func TestReadMissRehydratesFromCloud(t *testing.T) { // whole closure, so the install walk runs entirely locally — zero // per-effect network fetches. bc := &selectiveBroadcaster{ - allRegionPeersReachable: true, // in the majority partition - peerIDs: []pb.NodeID{7}, + mockBroadcaster: mockBroadcaster{ + allRegionPeersReachable: true, + peerIDs: []pb.NodeID{7}, + }, } e := newTestEngine(bc) fake := &fakeCloudReader{ @@ -155,9 +157,11 @@ func TestReadMissPartialSidecarFetchesStraggler(t *testing.T) { // The sidecar carries only the tip; the dep is the straggler, served by // the broadcaster. bc := &selectiveBroadcaster{ - allRegionPeersReachable: true, - peerIDs: []pb.NodeID{7}, - fetchable: map[Tip][]byte{depTip: depWire}, + mockBroadcaster: mockBroadcaster{ + allRegionPeersReachable: true, + peerIDs: []pb.NodeID{7}, + }, + fetchable: map[Tip][]byte{depTip: depWire}, } e := newTestEngine(bc) fake := &fakeCloudReader{ @@ -188,7 +192,9 @@ func TestReadMissPartialSidecarFetchesStraggler(t *testing.T) { // round-trip. func TestReadMissCloudEmptyStillMisses(t *testing.T) { bc := &selectiveBroadcaster{ - allRegionPeersReachable: true, + mockBroadcaster: mockBroadcaster{ + allRegionPeersReachable: true, + }, } e := newTestEngine(bc) fake := &fakeCloudReader{tips: map[string][]Tip{}} // Cloud holds nothing @@ -223,7 +229,9 @@ func TestReadMissCloudEmptyStillMisses(t *testing.T) { // back mid-outage must still rehydrate the key). func TestReadMissCloudErrorFailsRead(t *testing.T) { bc := &selectiveBroadcaster{ - allRegionPeersReachable: true, + mockBroadcaster: mockBroadcaster{ + allRegionPeersReachable: true, + }, } e := newTestEngine(bc) fake := &fakeCloudReader{err: context.DeadlineExceeded} @@ -268,9 +276,11 @@ func TestReadMissServesReadableSubsetAndMarksPending(t *testing.T) { holed := Tip{9, 4} // blob provably absent from cloud storage bc := &partialHoleBroadcaster{ - allRegionPeersReachable: true, - peerIDs: []pb.NodeID{7}, - fetchable: map[Tip][]byte{reachable: wire}, + mockBroadcaster: mockBroadcaster{ + allRegionPeersReachable: true, + peerIDs: []pb.NodeID{7}, + }, + fetchable: map[Tip][]byte{reachable: wire}, } e := newTestEngine(bc) fake := &fakeCloudReader{ @@ -310,7 +320,11 @@ func TestReadMissAllTipsPendingFailsRead(t *testing.T) { const key = "evicted" holed := Tip{9, 4} - bc := &holedFetchBroadcaster{allRegionPeersReachable: true} + bc := &holedFetchBroadcaster{ + mockBroadcaster: mockBroadcaster{ + allRegionPeersReachable: true, + }, + } e := newTestEngine(bc) fake := &fakeCloudReader{ tips: map[string][]Tip{key: {holed}}, diff --git a/effects/nack_ghost_test.go b/effects/nack_ghost_test.go index 415cbe8..d68b7e0 100644 --- a/effects/nack_ghost_test.go +++ b/effects/nack_ghost_test.go @@ -188,9 +188,11 @@ func TestEnsureSubscribed_InstallsOnlyWalkableTips(t *testing.T) { ghostTip := Tip{99999, 4} bc := &selectiveBroadcaster{ - peerIDs: []pb.NodeID{peerNode}, - allRegionPeersReachable: true, - nackTips: []*pb.EffectRef{toPbRef(realTip), toPbRef(ghostTip)}, + mockBroadcaster: mockBroadcaster{ + peerIDs: []pb.NodeID{peerNode}, + allRegionPeersReachable: true, + }, + nackTips: []*pb.EffectRef{toPbRef(realTip), toPbRef(ghostTip)}, fetchable: map[Tip][]byte{ realTip: realBytes, // ghostTip intentionally absent — FetchFromAny returns error. @@ -255,10 +257,12 @@ func TestEnsureSubscribed_AllTipsUnreachable_RetriesBootstrap(t *testing.T) { ghostB := Tip{500, 2} bc := &selectiveBroadcaster{ - peerIDs: []pb.NodeID{peerNode}, - allRegionPeersReachable: true, - nackTips: []*pb.EffectRef{toPbRef(ghostA), toPbRef(ghostB)}, - fetchable: map[Tip][]byte{}, // nothing fetchable + mockBroadcaster: mockBroadcaster{ + peerIDs: []pb.NodeID{peerNode}, + allRegionPeersReachable: true, + }, + nackTips: []*pb.EffectRef{toPbRef(ghostA), toPbRef(ghostB)}, + fetchable: map[Tip][]byte{}, // nothing fetchable } e := newTestEngine(bc) @@ -416,10 +420,12 @@ func TestRetryBootstrap_ShutdownUnblocksWaiters(t *testing.T) { const peerNode pb.NodeID = 7 bc := &selectiveBroadcaster{ - peerIDs: []pb.NodeID{peerNode}, - allRegionPeersReachable: true, - nackTips: []*pb.EffectRef{toPbRef(Tip{500, 1})}, - fetchable: map[Tip][]byte{}, + mockBroadcaster: mockBroadcaster{ + peerIDs: []pb.NodeID{peerNode}, + allRegionPeersReachable: true, + }, + nackTips: []*pb.EffectRef{toPbRef(Tip{500, 1})}, + fetchable: map[Tip][]byte{}, } e := newTestEngine(bc) diff --git a/effects/snapshot_test.go b/effects/snapshot_test.go index 1b3ccdf..1d35d07 100644 --- a/effects/snapshot_test.go +++ b/effects/snapshot_test.go @@ -1195,9 +1195,11 @@ func TestSubscriptionBootstrap_FetchesRemoteState(t *testing.T) { // Set up "local" engine (node 1) with no data localLog := newSnapshotLog() - // Create the bootstrap broadcaster that connects local and remote bc := &bootstrapBroadcaster{ - peerIDs: []pb.NodeID{2}, allRegionPeersReachable: true, + mockBroadcaster: mockBroadcaster{ + peerIDs: []pb.NodeID{2}, + allRegionPeersReachable: true, + }, } // The remote engine's broadcaster needs to forward NACKs to the local engine // We'll set this up after creating the local engine diff --git a/go.mod b/go.mod index 1a8cd0b..04e8387 100644 --- a/go.mod +++ b/go.mod @@ -1,10 +1,10 @@ module github.com/swytchdb/engine -go 1.27.0 +go 1.26.5 require ( github.com/cespare/xxhash/v2 v2.3.0 - github.com/klauspost/compress v1.19.2 + github.com/klauspost/compress v1.20.0 github.com/prometheus/client_golang v1.24.1 github.com/prometheus/client_model v0.6.3 github.com/puzpuzpuz/xsync/v4 v4.5.0 @@ -14,7 +14,7 @@ require ( go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.46.0 go.opentelemetry.io/otel/sdk v1.46.0 go.opentelemetry.io/otel/trace v1.46.0 - golang.org/x/crypto v0.55.0 + golang.org/x/crypto v0.56.0 golang.org/x/sync v0.22.0 google.golang.org/grpc v1.83.2 google.golang.org/protobuf v1.36.12 diff --git a/go.sum b/go.sum index 259461f..1d4a7fd 100644 --- a/go.sum +++ b/go.sum @@ -19,6 +19,8 @@ github.com/grpc-ecosystem/grpc-gateway/v2 v2.30.0 h1:/Tnpcb2E0Pz/tN9s3bfEY2Q8ePC github.com/grpc-ecosystem/grpc-gateway/v2 v2.30.0/go.mod h1:zOBXOsUaBSjKgmH4OGzV1esUpR3oUSCPYVd2cUBjKYY= github.com/klauspost/compress v1.19.2 h1:hMRETovs/pu/dVWN7zIT1PGG8t509MwT6bO7XSi26R8= github.com/klauspost/compress v1.19.2/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA= +github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= github.com/klauspost/cpuid/v2 v2.4.0 h1:S6Hrbc7+ywsr0r+RLapfGBHfyefhCTwEh3A0tV913Dw= github.com/klauspost/cpuid/v2 v2.4.0/go.mod h1:19jmZ9mjzoF//ddRSUsv0zfBTJWh3QJh9FNxZTMrGxU= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= @@ -71,6 +73,8 @@ go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw= go.yaml.in/yaml/v3 v3.0.5/go.mod h1:HVTZu1O7/Vkt2N+BFy8Zza+lnLsABggaTM2ZpNIGuKg= golang.org/x/crypto v0.55.0 h1:+KWHjbgOaAQ66dh/YlkZKHlz9ZUlq61AFirAR9ntP8M= golang.org/x/crypto v0.55.0/go.mod h1:uq0V9dE/fzQuJtbnL+2EhWOE63vo164FY8xqEnV9xis= +golang.org/x/crypto v0.56.0 h1:GUh5Ii4J5jtcseSMiRqr1jXCNHoxjeV9Fmekc2oLy6Y= +golang.org/x/crypto v0.56.0/go.mod h1:OMW5y6CY9l38uPLmxU6l6pwcXp1obtLo3e6gT7gQR2I= golang.org/x/net v0.58.0 h1:ynWG7rqYi4ccpTEuPZ2QGWHktVEM9DMCj9yzDE0Q7To= golang.org/x/net v0.58.0/go.mod h1:YwCddHnFlT7eLQqVprV19OnhLGtc5xOKgE0RyqgfWAU= golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek=