diff --git a/vermeer/algorithms/betweenness_centrality.go b/vermeer/algorithms/betweenness_centrality.go
index f8b28535c..3aee012db 100644
--- a/vermeer/algorithms/betweenness_centrality.go
+++ b/vermeer/algorithms/betweenness_centrality.go
@@ -18,8 +18,10 @@ under the License.
 package algorithms
 
 import (
+	"encoding/binary"
 	"fmt"
-	"math/rand"
+	"math"
+	"sync"
 	"vermeer/apps/common"
 	"vermeer/apps/compute"
 	"vermeer/apps/options"
@@ -28,6 +30,17 @@ import (
 	"github.com/sirupsen/logrus"
 )
 
+const (
+	betweennessStopValue     = "stop_sum"
+	betweennessCompleteValue = "betweenness_complete"
+
+	// A contribution chunk stays well below the 2 MiB scatter buffer. Several
+	// chunks can therefore be sent in one bounded RPC frame, while no result ever
+	// becomes a whole-graph compute value or a single multi-gigabyte message.
+	maxBetweennessChunkRecords = 65_536
+	betweennessRecordSize      = 8 // uint32 target + float32 worker partial
+)
+
 func init() {
 	Algorithms = append(Algorithms, new(BetweennessCentralityMaker))
 }
@@ -52,176 +65,536 @@ func (bc *BetweennessCentralityMaker) DataNeeded() []string {
 	return []string{}
 }
 
+func validateBetweennessSampleRate(value float64) error {
+	if math.IsNaN(value) || math.IsInf(value, 0) || value <= 0 || value > 1 {
+		return fmt.Errorf("betweenness_centrality.sample_rate must be finite and in (0, 1], got %v",
+			value)
+	}
+	return nil
+}
+
+func validateBetweennessMaxStep(maxStep int) error {
+	if maxStep < 2 {
+		return fmt.Errorf("betweenness_centrality requires compute.max_step >= 2, got %d",
+			maxStep)
+	}
+	return nil
+}
+
+func parseBetweennessEndpoint(value int) (bool, error) {
+	switch value {
+	case 0:
+		return false, nil
+	case 1:
+		return true, nil
+	default:
+		return false, fmt.Errorf("betweenness_centrality.use_endpoint must be 0 or 1, got %d",
+			value)
+	}
+}
+
+func betweennessScale(totalVertices uint32, useEndpoint bool, sampleRate float64) float64 {
+	if totalVertices <= 1 {
+		return 0
+	}
+	n := float64(totalVertices)
+	var scale float64
+	if useEndpoint {
+		scale = 1 / (n * (n - 1))
+	} else {
+		if totalVertices <= 2 {
+			return 0
+		}
+		scale = 1 / ((n - 1) * (n - 2))
+	}
+	return scale / sampleRate
+}
+
+func sampleBetweennessSource(vertexID uint32, sampleRate float64) bool {
+	if sampleRate >= 1 {
+		return true
+	}
+	value := uint64(vertexID) + 0x9e3779b97f4a7c15
+	value = (value ^ (value >> 30)) * 0xbf58476d1ce4e5b9
+	value = (value ^ (value >> 27)) * 0x94d049bb133111eb
+	value ^= value >> 31
+	return float64(value>>11) < sampleRate*float64(uint64(1)<<53)
+}
+
+func selectBetweennessSources(start, count uint32, sampleRate float64) []uint32 {
+	if count == 0 {
+		return nil
+	}
+	capacity := int(math.Ceil(float64(count) * sampleRate))
+	if capacity < 1 {
+		capacity = 1
+	}
+	sources := make([]uint32, 0, capacity)
+	for localID := uint32(0); localID < count; localID++ {
+		vertexID := start + localID
+		if sampleBetweennessSource(vertexID, sampleRate) {
+			sources = append(sources, vertexID)
+		}
+	}
+	return sources
+}
+
+func logAddExp(left, right float64) float64 {
+	if math.IsInf(left, -1) {
+		return right
+	}
+	if math.IsInf(right, -1) {
+		return left
+	}
+	if left < right {
+		left, right = right, left
+	}
+	return left + math.Log1p(math.Exp(right-left))
+}
+
+func brandesContributions(source uint32, neighbors []serialize.SliceUint32,
+	useEndpoint bool, scale float64) map[uint32]float64 {
+	result := make(map[uint32]float64)
+	if scale == 0 || int(source) >= len(neighbors) {
+		return result
+	}
+
+	predecessors := make(map[uint32][]uint32)
+	logSigma := map[uint32]float64{source: 0}
+	distance := map[uint32]int{source: 0}
+	queue := []uint32{source}
+
+	for head := 0; head < len(queue); head++ {
+		vertex := queue[head]
+		nextDistance := distance[vertex] + 1
+		for _, rawNeighbor := range neighbors[vertex] {
+			neighbor := uint32(rawNeighbor)
+			if int(neighbor) >= len(neighbors) {
+				continue
+			}
+			neighborDistance, seen := distance[neighbor]
+			if !seen {
+				neighborDistance = nextDistance
+				distance[neighbor] = neighborDistance
+				queue = append(queue, neighbor)
+			}
+			if neighborDistance != nextDistance {
+				continue
+			}
+			currentLogSigma, ok := logSigma[neighbor]
+			if !ok {
+				currentLogSigma = math.Inf(-1)
+			}
+			logSigma[neighbor] = logAddExp(currentLogSigma, logSigma[vertex])
+			predecessors[neighbor] = append(predecessors[neighbor], vertex)
+		}
+	}
+
+	dependency := make(map[uint32]float64, len(queue))
+	if useEndpoint && len(queue) > 1 {
+		result[source] = float64(len(queue)-1) * scale
+	}
+	for index := len(queue) - 1; index >= 0; index-- {
+		vertex := queue[index]
+		for _, predecessor := range predecessors[vertex] {
+			pathRatio := math.Exp(logSigma[predecessor] - logSigma[vertex])
+			dependency[predecessor] += pathRatio * (1 + dependency[vertex])
+		}
+		if vertex == source {
+			continue
+		}
+		contribution := dependency[vertex]
+		if useEndpoint {
+			contribution++
+		}
+		contribution *= scale
+		if contribution != 0 {
+			result[vertex] = contribution
+		}
+	}
+	return result
+}
+
+// betweennessContributionChunk is already encoded so its retained memory is
+// exactly the wire representation: a uint32 count followed by count pairs of
+// (uint32 target, float32 worker partial). It never contains a whole-graph
+// vector and each chunk has a fixed upper bound.
+type betweennessContributionChunk struct {
+	payload []byte
+}
+
+func (chunk *betweennessContributionChunk) Marshal(buffer []byte) (int, error) {
+	if len(buffer) < len(chunk.payload) {
+		return 0, fmt.Errorf("betweenness contribution buffer has %d bytes, need %d",
+			len(buffer), len(chunk.payload))
+	}
+	return copy(buffer, chunk.payload), nil
+}
+
+func (chunk *betweennessContributionChunk) Unmarshal(buffer []byte) (int, error) {
+	size, err := betweennessChunkSize(buffer)
+	if err != nil {
+		return 0, err
+	}
+	chunk.payload = append(chunk.payload[:0], buffer[:size]...)
+	return size, nil
+}
+
+func (chunk *betweennessContributionChunk) ToString() string {
+	if len(chunk.payload) < 4 {
+		return "0"
+	}
+	return fmt.Sprintf("%d", binary.LittleEndian.Uint32(chunk.payload[:4]))
+}
+
+func (chunk *betweennessContributionChunk) PredictSize() int {
+	return len(chunk.payload)
+}
+
+func betweennessChunkSize(buffer []byte) (int, error) {
+	if len(buffer) < 4 {
+		return 0, fmt.Errorf("betweenness contribution chunk is missing its count")
+	}
+	count := uint64(binary.LittleEndian.Uint32(buffer[:4]))
+	size := uint64(4) + count*betweennessRecordSize
+	if size > uint64(len(buffer)) {
+		return 0, fmt.Errorf("truncated betweenness contribution chunk: need %d bytes, got %d",
+			size, len(buffer))
+	}
+	return int(size), nil
+}
+
+func appendBetweennessRecord(payload []byte, target uint32, value float32) []byte {
+	offset := len(payload)
+	payload = append(payload, make([]byte, betweennessRecordSize)...)
+	binary.LittleEndian.PutUint32(payload[offset:offset+4], target)
+	binary.LittleEndian.PutUint32(payload[offset+4:offset+8], math.Float32bits(value))
+	return payload
+}
+
+func splitBetweennessContributions(partial []float64, localStart,
+	localCount uint32) ([]float64, []betweennessContributionChunk, error) {
+	local := make([]float64, localCount)
+	localEnd := uint64(localStart) + uint64(localCount)
+	chunks := make([]betweennessContributionChunk, 0)
+	var payload []byte
+	records := 0
+
+	flush := func() {
+		if records == 0 {
+			return
+		}
+		binary.LittleEndian.PutUint32(payload[:4], uint32(records))
+		chunks = append(chunks, betweennessContributionChunk{payload: payload})
+		payload = nil
+		records = 0
+	}
+
+	for target, value := range partial {
+		if value == 0 {
+			continue
+		}
+		if math.IsNaN(value) || math.IsInf(value, 0) {
+			return nil, nil, fmt.Errorf("betweenness partial for vertex %d is not finite", target)
+		}
+		rounded := float32(value)
+		if math.IsInf(float64(rounded), 0) {
+			return nil, nil, fmt.Errorf("betweenness partial for vertex %d exceeds float32", target)
+		}
+		if rounded == 0 {
+			continue
+		}
+
+		targetID := uint32(target)
+		if uint64(targetID) >= uint64(localStart) && uint64(targetID) < localEnd {
+			local[targetID-localStart] += float64(rounded)
+			continue
+		}
+		if records == 0 {
+			payload = make([]byte, 4, 4+maxBetweennessChunkRecords*betweennessRecordSize)
+		}
+		payload = appendBetweennessRecord(payload, targetID, rounded)
+		records++
+		if records == maxBetweennessChunkRecords {
+			flush()
+		}
+	}
+	flush()
+	return local, chunks, nil
+}
+
+type betweennessContributionReceiver struct {
+	worker *BetweennessCentralityWorker
+}
+
+func (*betweennessContributionReceiver) Marshal([]byte) (int, error) {
+	// Inactive local vertex slots intentionally contribute no bytes. Active
+	// slots return a betweennessContributionChunk from VertexValue instead.
+	return 0, nil
+}
+
+func (receiver *betweennessContributionReceiver) Unmarshal(buffer []byte) (int, error) {
+	if receiver.worker == nil {
+		return 0, fmt.Errorf("betweenness contribution receiver is not initialized")
+	}
+	size, err := betweennessChunkSize(buffer)
+	if err != nil {
+		return 0, err
+	}
+	count := int(binary.LittleEndian.Uint32(buffer[:4]))
+	worker := receiver.worker
+	localEnd := uint64(worker.localStart) + uint64(worker.localCount)
+
+	worker.valuesMu.Lock()
+	defer worker.valuesMu.Unlock()
+	for index := 0; index < count; index++ {
+		offset := 4 + index*betweennessRecordSize
+		target := binary.LittleEndian.Uint32(buffer[offset : offset+4])
+		value := math.Float32frombits(binary.LittleEndian.Uint32(
+			buffer[offset+4 : offset+8]))
+		if math.IsNaN(float64(value)) || math.IsInf(float64(value), 0) {
+			return 0, fmt.Errorf("received non-finite betweenness value for vertex %d", target)
+		}
+		if uint64(target) < uint64(worker.localStart) || uint64(target) >= localEnd {
+			continue
+		}
+		worker.localValues[target-worker.localStart] += float64(value)
+	}
+	return size, nil
+}
+
+func (*betweennessContributionReceiver) ToString() string {
+	return ""
+}
+
+func (*betweennessContributionReceiver) PredictSize() int {
+	return 0
+}
+
 type BetweennessCentralityWorker struct {
 	compute.WorkerComputerBase
-	useEndPoint         bool
-	stopFlag            bool
-	scale               serialize.SFloat32
-	sampleRate          float64
-	stopSum             []serialize.SInt32
-	betweennessValues   []serialize.SFloat32
-	tempEdges           [][]serialize.SUint32
-	neighborEdges       []serialize.SliceUint32
-	betweennessValueMap []serialize.MapUint32Float32
-	bfsNewValues        []map[uint32][][]uint32
-	bfsOldValues        []map[uint32][][]uint32
-}
-
-func (bcw *BetweennessCentralityWorker) VertexValue(i uint32) serialize.MarshalAble {
+	useEndPoint       bool
+	scale             float64
+	sampleRate        float64
+	totalVertices     uint32
+	localStart        uint32
+	localCount        uint32
+	sampledSources    []uint32
+	allSources        bool
+	selectedSources   int
+	neighborEdges     []serialize.SliceUint32
+	localValues       []float64
+	betweennessValues serialize.SliceFloat32
+	outgoingChunks    []betweennessContributionChunk
+	outgoingOffset    int
+	computed          bool
+	valuesMu          sync.Mutex
+	receiver          betweennessContributionReceiver
+}
+
+func (bcw *BetweennessCentralityWorker) VertexValue(vertexID uint32) serialize.MarshalAble {
+	if bcw.WContext.Output {
+		if vertexID < bcw.localStart || vertexID >= bcw.localStart+bcw.localCount {
+			return &bcw.receiver
+		}
+		return &bcw.betweennessValues[vertexID-bcw.localStart]
+	}
 	if bcw.WContext.Step == 1 {
-		return &bcw.neighborEdges[i]
+		return &bcw.neighborEdges[vertexID]
+	}
+
+	// The standard scatter loop provides one slot per local vertex. Fill the
+	// current window with bounded contribution chunks and leave every remaining
+	// slot at zero bytes. Receivers do not rely on the slot's vertex ID because
+	// each record carries its actual target ID.
+	if vertexID >= bcw.localStart && vertexID < bcw.localStart+bcw.localCount {
+		localID := int(vertexID - bcw.localStart)
+		window := bcw.outgoingWindowCount()
+		if localID < window {
+			return &bcw.outgoingChunks[bcw.outgoingOffset+localID]
+		}
 	}
-	if bcw.stopFlag && bcw.WContext.Step > 1 {
-		return &bcw.betweennessValues[i]
+	return &bcw.receiver
+}
+
+func (bcw *BetweennessCentralityWorker) ValidateOutput() error {
+	cValue, ok := bcw.WContext.CValues[betweennessCompleteValue]
+	if !ok || cValue.Value == nil {
+		return fmt.Errorf("betweenness_centrality output is missing completion state")
+	}
+	complete, ok := cValue.Value.(serialize.SInt32)
+	if !ok || complete != 1 {
+		return fmt.Errorf("betweenness_centrality did not finish contribution scattering; increase compute.max_step")
+	}
+	if uint32(len(bcw.localValues)) != bcw.localCount {
+		return fmt.Errorf("betweenness_centrality local result has %d values, expected %d",
+			len(bcw.localValues), bcw.localCount)
+	}
+	bcw.betweennessValues = make(serialize.SliceFloat32, bcw.localCount)
+	for localID, value := range bcw.localValues {
+		if math.IsNaN(value) || math.IsInf(value, 0) {
+			return fmt.Errorf("betweenness result for local vertex %d is not finite", localID)
+		}
+		bcw.betweennessValues[localID] = serialize.SFloat32(value)
 	}
-	return &bcw.betweennessValueMap[i]
+	return nil
 }
 
 func (bcw *BetweennessCentralityWorker) Init() error {
-	bcw.betweennessValues = make([]serialize.SFloat32, bcw.WContext.GraphData.Vertex.TotalVertexCount())
-	bcw.neighborEdges = make([]serialize.SliceUint32, bcw.WContext.GraphData.Vertex.TotalVertexCount())
-	bcw.betweennessValueMap = make([]serialize.MapUint32Float32, bcw.WContext.GraphData.Vertex.TotalVertexCount())
-	for i := range bcw.betweennessValueMap {
-		bcw.betweennessValueMap[i] = make(map[serialize.SUint32]serialize.SFloat32)
-	}
-	bcw.tempEdges = make([][]serialize.SUint32, bcw.WContext.Parallel)
-	for i := range bcw.tempEdges {
-		bcw.tempEdges[i] = make([]serialize.SUint32, 0, 100)
-	}
-	bcw.bfsNewValues = make([]map[uint32][][]uint32, bcw.WContext.GraphData.VertexCount)
-	bcw.bfsOldValues = make([]map[uint32][][]uint32, bcw.WContext.GraphData.VertexCount)
-	bcw.stopSum = make([]serialize.SInt32, bcw.WContext.Parallel)
-	bcw.sampleRate = options.GetFloat(bcw.WContext.Params, "betweenness_centrality.sample_rate")
-	if bcw.sampleRate > 1.0 || bcw.sampleRate <= 0 {
-		logrus.Errorf("The param betweenness_centrality.sample_rate must be in (0.0, 1.0], actual got '%v",
-			bcw.sampleRate)
-		return fmt.Errorf("the param betweenness_centrality.sample_rate must be in (0.0, 1.0], actual got '%v",
-			bcw.sampleRate)
+	bcw.sampleRate = options.GetFloat(bcw.WContext.Params,
+		"betweenness_centrality.sample_rate")
+	if err := validateBetweennessSampleRate(bcw.sampleRate); err != nil {
+		return err
+	}
+	useEndpoint, err := parseBetweennessEndpoint(options.GetInt(bcw.WContext.Params,
+		"betweenness_centrality.use_endpoint"))
+	if err != nil {
+		return err
 	}
-	bcw.useEndPoint = false
-	if options.GetInt(bcw.WContext.Params, "betweenness_centrality.use_endpoint") == 1 {
-		bcw.useEndPoint = true
+	if err := validateBetweennessMaxStep(options.GetInt(bcw.WContext.Params,
+		"compute.max_step")); err != nil {
+		return err
 	}
-	bcw.stopFlag = false
 
-	n := serialize.SFloat32(bcw.WContext.GraphData.Vertex.TotalVertexCount())
-	if n < 2 {
-		bcw.scale = 1
-	} else if bcw.useEndPoint {
-		n *= serialize.SFloat32(bcw.sampleRate)
-		bcw.scale = 1 / serialize.SFloat32(n*(n-1))
+	bcw.useEndPoint = useEndpoint
+	bcw.totalVertices = bcw.WContext.GraphData.Vertex.TotalVertexCount()
+	bcw.localStart = bcw.WContext.GraphData.VertIDStart
+	bcw.localCount = bcw.WContext.GraphData.VertexCount
+	bcw.scale = betweennessScale(bcw.totalVertices, bcw.useEndPoint, bcw.sampleRate)
+	if bcw.sampleRate >= 1 {
+		bcw.allSources = true
+		bcw.selectedSources = int(bcw.localCount)
 	} else {
-		n *= serialize.SFloat32(bcw.sampleRate)
-		bcw.scale = 1 / serialize.SFloat32((n-1)*(n-2))
+		bcw.sampledSources = selectBetweennessSources(bcw.localStart, bcw.localCount,
+			bcw.sampleRate)
+		bcw.selectedSources = len(bcw.sampledSources)
 	}
+	bcw.neighborEdges = make([]serialize.SliceUint32, bcw.totalVertices)
+	bcw.localValues = make([]float64, bcw.localCount)
+	bcw.betweennessValues = make(serialize.SliceFloat32, bcw.localCount)
+	bcw.receiver.worker = bcw
 
-	bcw.WContext.CreateValue("stop_sum", compute.ValueTypeInt32, compute.CValueActionAggregate)
-	bcw.WContext.SetValue("stop_sum", serialize.SInt32(0))
-	bcw.WContext.CreateValue("stop_flag", compute.ValueTypeInt32, compute.CValueActionAggregate)
-	bcw.WContext.SetValue("stop_flag", serialize.SInt32(0))
+	bcw.WContext.CreateValue(betweennessStopValue, compute.ValueTypeInt32,
+		compute.CValueActionAggregate)
+	bcw.WContext.SetValue(betweennessStopValue, serialize.SInt32(1))
+	bcw.WContext.CreateValue(betweennessCompleteValue, compute.ValueTypeInt32,
+		compute.CValueActionNormal)
+	bcw.WContext.SetValue(betweennessCompleteValue, serialize.SInt32(0))
 	return nil
 }
 
 func (bcw *BetweennessCentralityWorker) BeforeStep() {
-	for i := range bcw.stopSum {
-		bcw.stopSum[i] = 0
-	}
-	if bcw.WContext.Step > 1 && bcw.WContext.GetValue("stop_flag").(serialize.SInt32) == 1 ||
-		options.GetInt(bcw.WContext.Params, "compute.max_step") == int(bcw.WContext.Step) {
-		bcw.stopFlag = true
-	}
-	for i := range bcw.bfsNewValues {
-		bcw.bfsOldValues[i] = make(map[uint32][][]uint32)
-		for k, v := range bcw.bfsNewValues[i] {
-			bcw.bfsOldValues[i][k] = v
-		}
-		bcw.bfsNewValues[i] = make(map[uint32][][]uint32)
-	}
-
-	if bcw.stopFlag {
-		rangeStart := bcw.WContext.GraphData.VertIDStart
-		rangeEnd := bcw.WContext.GraphData.VertIDStart + bcw.WContext.GraphData.VertexCount
-		for _, valueMap := range bcw.betweennessValueMap {
-			for k, v := range valueMap {
-				key := uint32(k)
-				if key >= rangeStart && key < rangeEnd {
-					bcw.betweennessValues[key] += v
-				}
-			}
-		}
+	if bcw.WContext.Step != 2 || bcw.computed {
+		return
 	}
-
+	partial := bcw.computeSampledSourcePartials()
+	local, chunks, err := splitBetweennessContributions(partial,
+		bcw.localStart, bcw.localCount)
+	if err != nil {
+		panic(err)
+	}
+	bcw.localValues = local
+	bcw.outgoingChunks = chunks
+	bcw.outgoingOffset = 0
+	bcw.computed = true
+	logrus.Infof("betweenness selected sources: %d, contribution chunks: %d",
+		bcw.selectedSources, len(bcw.outgoingChunks))
 }
 
-func (bcw *BetweennessCentralityWorker) Compute(vertexID uint32, pID int) {
-	if bcw.stopFlag {
+func (bcw *BetweennessCentralityWorker) Compute(vertexID uint32, _ int) {
+	if bcw.WContext.Step != 1 {
 		return
 	}
-	vertID := vertexID - bcw.WContext.GraphData.VertIDStart
-	switch bcw.WContext.Step {
-	case 1:
-		inEdges := bcw.WContext.GraphData.Edges.GetInEdges(vertID)
-		bcw.neighborEdges[vertexID] = make([]serialize.SUint32, 0, len(inEdges))
-		for _, edge := range inEdges {
-			if !bcw.sample() || uint32(edge) == vertexID {
-				continue
-			}
-			bcw.neighborEdges[vertexID] = append(bcw.neighborEdges[vertexID], edge)
-			bcw.bfsNewValues[vertID][uint32(edge)] = [][]uint32{}
-			bcw.betweennessValueMap[vertexID][edge] = 0
-			if bcw.useEndPoint {
-				bcw.betweennessValueMap[vertexID][edge] += 1 * bcw.scale
-			}
+	localID := vertexID - bcw.localStart
+	inEdges := bcw.WContext.GraphData.Edges.GetInEdges(localID)
+	neighbors := make(serialize.SliceUint32, 0, len(inEdges))
+	seen := make(map[serialize.SUint32]struct{}, len(inEdges))
+	for _, neighbor := range inEdges {
+		if uint32(neighbor) == vertexID || uint32(neighbor) >= bcw.totalVertices {
+			continue
 		}
-		bcw.stopSum[pID] = 1
-	default:
-		for k, v := range bcw.bfsOldValues[vertID] {
-			for _, nId := range bcw.neighborEdges[k] {
-				if !bcw.sample() || nId == serialize.SUint32(vertexID) {
-					continue
-				}
-				if _, ok := bcw.betweennessValueMap[vertexID][nId]; !ok {
-					bcw.stopSum[pID] = 1
-					bcw.tempEdges[pID] = append(bcw.tempEdges[pID], nId)
-					for _, vi := range v {
-						newV := append(vi, k)
-						bcw.bfsNewValues[vertID][uint32(nId)] = append(bcw.bfsNewValues[vertID][uint32(nId)], newV)
-					}
-					if len(v) == 0 {
-						bcw.bfsNewValues[vertID][uint32(nId)] = append(bcw.bfsNewValues[vertID][uint32(nId)], []uint32{k})
-					}
-				}
-			}
-		}
-		for _, edge := range bcw.tempEdges[pID] {
-			bcw.betweennessValueMap[vertexID][edge] = 0
-			if bcw.useEndPoint {
-				bcw.betweennessValueMap[vertexID][edge] += 1 * bcw.scale
-			}
+		if _, exists := seen[neighbor]; exists {
+			continue
 		}
+		seen[neighbor] = struct{}{}
+		neighbors = append(neighbors, neighbor)
 	}
-	bcw.tempEdges[pID] = bcw.tempEdges[pID][0:0]
-	for _, values := range bcw.bfsNewValues[vertID] {
-		for _, v := range values {
-			for _, u := range v {
-				bcw.betweennessValueMap[vertexID][serialize.SUint32(u)] +=
-					1.0 / serialize.SFloat32(len(values)) * bcw.scale
-			}
+	bcw.neighborEdges[vertexID] = neighbors
+}
+
+func (bcw *BetweennessCentralityWorker) AfterStep() {
+	if bcw.WContext.Step == 1 {
+		if bcw.selectedSources == 0 || bcw.scale == 0 {
+			bcw.WContext.SetValue(betweennessStopValue, serialize.SInt32(0))
+		} else {
+			bcw.WContext.SetValue(betweennessStopValue, serialize.SInt32(1))
 		}
+		return
 	}
 
+	bcw.outgoingOffset += bcw.outgoingWindowCount()
+	if bcw.outgoingOffset < len(bcw.outgoingChunks) {
+		bcw.WContext.SetValue(betweennessStopValue, serialize.SInt32(1))
+		return
+	}
+	bcw.outgoingChunks = nil
+	bcw.WContext.SetValue(betweennessStopValue, serialize.SInt32(0))
 }
 
-func (bcw *BetweennessCentralityWorker) AfterStep() {
-	var stopSum serialize.SInt32
-	for i := range bcw.stopSum {
-		stopSum += bcw.stopSum[i]
+func (bcw *BetweennessCentralityWorker) outgoingWindowCount() int {
+	remaining := len(bcw.outgoingChunks) - bcw.outgoingOffset
+	if remaining <= 0 || bcw.localCount == 0 {
+		return 0
+	}
+	if remaining < int(bcw.localCount) {
+		return remaining
+	}
+	return int(bcw.localCount)
+}
+
+func (bcw *BetweennessCentralityWorker) computeSampledSourcePartials() []float64 {
+	partial := make([]float64, bcw.totalVertices)
+	if bcw.selectedSources == 0 || bcw.scale == 0 {
+		return partial
+	}
+	parallel := bcw.WContext.Parallel
+	if parallel <= 0 {
+		parallel = 1
+	}
+	if parallel > bcw.selectedSources {
+		parallel = bcw.selectedSources
+	}
+
+	results := make(chan map[uint32]float64, parallel)
+	var workers sync.WaitGroup
+	workers.Add(parallel)
+	for workerID := 0; workerID < parallel; workerID++ {
+		go func(offset int) {
+			defer workers.Done()
+			for sourceIndex := offset; sourceIndex < bcw.selectedSources; sourceIndex += parallel {
+				results <- brandesContributions(bcw.selectedSource(sourceIndex),
+					bcw.neighborEdges, bcw.useEndPoint, bcw.scale)
+			}
+		}(workerID)
 	}
-	bcw.WContext.SetValue("stop_sum", stopSum)
+	go func() {
+		workers.Wait()
+		close(results)
+	}()
+
+	for contributions := range results {
+		for target, value := range contributions {
+			partial[target] += value
+		}
+	}
+	return partial
 }
 
-func (bcw *BetweennessCentralityWorker) sample() bool {
-	return rand.Float64() < bcw.sampleRate
+func (bcw *BetweennessCentralityWorker) selectedSource(index int) uint32 {
+	if bcw.allSources {
+		return bcw.localStart + uint32(index)
+	}
+	return bcw.sampledSources[index]
 }
 
 func (bcw *BetweennessCentralityWorker) OutputValueType() string {
@@ -232,15 +605,22 @@ type BetweennessCentralityMaster struct {
 	compute.MasterComputerBase
 }
 
-func (bcm *BetweennessCentralityMaster) Compute() bool {
-	logrus.Infof("stopFlag:%v", bcm.MContext.GetValue("stop_flag").(serialize.SInt32))
-	if bcm.MContext.GetValue("stop_flag").(serialize.SInt32) == serialize.SInt32(len(bcm.MContext.WorkerCValues)) {
-		return false
+func (bcm *BetweennessCentralityMaster) Init() error {
+	if err := validateBetweennessMaxStep(options.GetInt(bcm.MContext.Params,
+		"compute.max_step")); err != nil {
+		return err
 	}
-	stopSum := bcm.MContext.GetValue("stop_sum")
-	logrus.Infof("stopSum:%v", stopSum)
-	if stopSum.(serialize.SInt32) == serialize.SInt32(0) {
-		bcm.MContext.SetValue("stop_flag", serialize.SInt32(1))
+	bcm.MContext.CreateValue(betweennessCompleteValue, compute.ValueTypeInt32,
+		compute.CValueActionNormal)
+	bcm.MContext.SetValue(betweennessCompleteValue, serialize.SInt32(0))
+	return nil
+}
+
+func (bcm *BetweennessCentralityMaster) Compute() bool {
+	stopSum := bcm.MContext.GetValue(betweennessStopValue).(serialize.SInt32)
+	if stopSum != 0 {
+		return true
 	}
-	return true
+	bcm.MContext.SetValue(betweennessCompleteValue, serialize.SInt32(1))
+	return false
 }
diff --git a/vermeer/algorithms/betweenness_centrality_test.go b/vermeer/algorithms/betweenness_centrality_test.go
new file mode 100644
index 000000000..fcced2076
--- /dev/null
+++ b/vermeer/algorithms/betweenness_centrality_test.go
@@ -0,0 +1,343 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with this
+work for additional information regarding copyright ownership. The ASF
+licenses this file to You under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package algorithms
+
+import (
+	"encoding/binary"
+	"math"
+	"testing"
+	"vermeer/apps/buffer"
+	"vermeer/apps/compute"
+	"vermeer/apps/serialize"
+)
+
+func TestBrandesBetweennessPath(t *testing.T) {
+	neighbors := []serialize.SliceUint32{{1}, {0, 2}, {1}}
+	scale := betweennessScale(3, false, 1)
+	values := make([]float64, 3)
+	for source := uint32(0); source < 3; source++ {
+		for vertex, value := range brandesContributions(source, neighbors, false, scale) {
+			values[vertex] += value
+		}
+	}
+	if math.Abs(values[1]-1) > 1e-6 {
+		t.Fatalf("middle vertex expected normalized betweenness 1, got %v", values[1])
+	}
+	if values[0] != 0 || values[2] != 0 {
+		t.Fatalf("path endpoints must have zero betweenness, got %v", values)
+	}
+}
+
+func TestBrandesSplitsEqualShortestPaths(t *testing.T) {
+	neighbors := []serialize.SliceUint32{{1, 2}, {3}, {3}, {}}
+	values := brandesContributions(0, neighbors, false, 1)
+	if math.Abs(values[1]-0.5) > 1e-6 ||
+		math.Abs(values[2]-0.5) > 1e-6 {
+		t.Fatalf("diamond paths were not split equally: %v", values)
+	}
+}
+
+func TestBrandesPathCountsStayFiniteBeyondFloat64Range(t *testing.T) {
+	const layers = 1030
+	neighbors := make([]serialize.SliceUint32, 1+layers*2)
+	neighbors[0] = serialize.SliceUint32{1, 2}
+	for layer := 0; layer < layers-1; layer++ {
+		left := uint32(1 + layer*2)
+		right := left + 1
+		nextLeft := left + 2
+		nextRight := left + 3
+		neighbors[left] = serialize.SliceUint32{serialize.SUint32(nextLeft),
+			serialize.SUint32(nextRight)}
+		neighbors[right] = serialize.SliceUint32{serialize.SUint32(nextLeft),
+			serialize.SUint32(nextRight)}
+	}
+
+	values := brandesContributions(0, neighbors, false, 1)
+	for vertex, value := range values {
+		if math.IsNaN(value) || math.IsInf(value, 0) {
+			t.Fatalf("vertex %d has non-finite contribution %v", vertex, value)
+		}
+	}
+	lastLeft := uint32(len(neighbors) - 2)
+	lastRight := lastLeft + 1
+	if math.Abs(values[lastLeft]-values[lastRight]) > 1e-9 {
+		t.Fatalf("symmetric final layer differs: left=%v right=%v",
+			values[lastLeft], values[lastRight])
+	}
+}
+
+func TestBrandesDisconnectedSourcesDoNotAllocateGlobalStacks(t *testing.T) {
+	const vertices = 4096
+	neighbors := make([]serialize.SliceUint32, vertices)
+	for source := uint32(0); source < vertices; source++ {
+		if values := brandesContributions(source, neighbors, false, 1); len(values) != 0 {
+			t.Fatalf("isolated source %d produced contributions: %v", source, values)
+		}
+	}
+}
+
+func TestBetweennessScaleAndValidation(t *testing.T) {
+	if scale := betweennessScale(2, false, 1); scale != 0 {
+		t.Fatalf("two-vertex non-endpoint normalization must be zero, got %v", scale)
+	}
+	if scale := betweennessScale(4, false, 0.5); math.Abs(scale-1.0/3.0) > 1e-12 {
+		t.Fatalf("unexpected sampled scale: %v", scale)
+	}
+	for _, value := range []float64{0, -1, 1.1, math.NaN(), math.Inf(1)} {
+		if err := validateBetweennessSampleRate(value); err == nil {
+			t.Fatalf("invalid sample rate %v accepted", value)
+		}
+	}
+	for _, value := range []int{-1, 2} {
+		if _, err := parseBetweennessEndpoint(value); err == nil {
+			t.Fatalf("invalid endpoint flag %d accepted", value)
+		}
+	}
+	if err := validateBetweennessMaxStep(1); err == nil {
+		t.Fatal("compute.max_step=1 was accepted")
+	}
+	if err := validateBetweennessMaxStep(2); err != nil {
+		t.Fatalf("compute.max_step=2 was rejected: %v", err)
+	}
+}
+
+func TestBetweennessSourceSamplingIsDeterministicAndCompacted(t *testing.T) {
+	first := selectBetweennessSources(100, 10_000, 0.01)
+	second := selectBetweennessSources(100, 10_000, 0.01)
+	if len(first) == 0 || len(first) >= 10_000 {
+		t.Fatalf("unexpected selected source count %d", len(first))
+	}
+	if len(first) != len(second) {
+		t.Fatalf("deterministic selections differ in length: %d vs %d",
+			len(first), len(second))
+	}
+	for index, source := range first {
+		if source != second[index] {
+			t.Fatalf("selection differs at %d: %d vs %d",
+				index, source, second[index])
+		}
+		if !sampleBetweennessSource(source, 0.01) {
+			t.Fatalf("unselected source %d was compacted into the work list", source)
+		}
+	}
+}
+
+func TestExactModeDoesNotRetainASecondSourceIDArray(t *testing.T) {
+	worker := &BetweennessCentralityWorker{
+		localStart:      50,
+		localCount:      3,
+		allSources:      true,
+		selectedSources: 3,
+	}
+	if len(worker.sampledSources) != 0 {
+		t.Fatalf("exact mode retained sampled source IDs: %v", worker.sampledSources)
+	}
+	for index, expected := range []uint32{50, 51, 52} {
+		if got := worker.selectedSource(index); got != expected {
+			t.Fatalf("source %d expected %d, got %d", index, expected, got)
+		}
+	}
+}
+
+func TestSplitBetweennessContributionsKeepsLocalAndChunksRemote(t *testing.T) {
+	partial := []float64{0, 1.25, 2.5, 3.75}
+	local, chunks, err := splitBetweennessContributions(partial, 1, 2)
+	if err != nil {
+		t.Fatalf("split contributions: %v", err)
+	}
+	if len(local) != 2 || local[0] != 1.25 || local[1] != 2.5 {
+		t.Fatalf("unexpected local contribution slice: %v", local)
+	}
+	if len(chunks) != 1 {
+		t.Fatalf("expected one remote chunk, got %d", len(chunks))
+	}
+	if count := binary.LittleEndian.Uint32(chunks[0].payload[:4]); count != 1 {
+		t.Fatalf("expected one remote record, got %d", count)
+	}
+
+	remote := &BetweennessCentralityWorker{
+		localStart:  3,
+		localCount:  1,
+		localValues: make([]float64, 1),
+	}
+	receiver := betweennessContributionReceiver{worker: remote}
+	if _, err := receiver.Unmarshal(chunks[0].payload); err != nil {
+		t.Fatalf("receive contribution chunk: %v", err)
+	}
+	if math.Abs(remote.localValues[0]-3.75) > 1e-6 {
+		t.Fatalf("remote contribution expected 3.75, got %v", remote.localValues[0])
+	}
+}
+
+func TestBetweennessContributionChunksStayBounded(t *testing.T) {
+	partial := make([]float64, maxBetweennessChunkRecords+5)
+	for index := range partial {
+		partial[index] = float64(index + 1)
+	}
+	_, chunks, err := splitBetweennessContributions(partial, 0, 0)
+	if err != nil {
+		t.Fatalf("split bounded chunks: %v", err)
+	}
+	if len(chunks) != 2 {
+		t.Fatalf("expected two chunks, got %d", len(chunks))
+	}
+	maximum := 4 + maxBetweennessChunkRecords*betweennessRecordSize
+	for index, chunk := range chunks {
+		if chunk.PredictSize() > maximum {
+			t.Fatalf("chunk %d has %d bytes, maximum is %d",
+				index, chunk.PredictSize(), maximum)
+		}
+		if chunk.PredictSize() >= 2*1024*1024 {
+			t.Fatalf("chunk %d is not scatter-buffer bounded: %d bytes",
+				index, chunk.PredictSize())
+		}
+	}
+}
+
+func TestBetweennessContributionChunkRejectsTruncation(t *testing.T) {
+	payload := make([]byte, 4)
+	binary.LittleEndian.PutUint32(payload, 1)
+	var chunk betweennessContributionChunk
+	if _, err := chunk.Unmarshal(payload); err == nil {
+		t.Fatal("truncated contribution chunk was accepted")
+	}
+
+	worker := &BetweennessCentralityWorker{
+		localCount:  1,
+		localValues: make([]float64, 1),
+	}
+	receiver := betweennessContributionReceiver{worker: worker}
+	if _, err := receiver.Unmarshal(payload); err == nil {
+		t.Fatal("receiver accepted a truncated contribution chunk")
+	}
+}
+
+func TestBetweennessInactiveScatterPayloadIsConstant(t *testing.T) {
+	var encoded buffer.EncodeBuffer
+	encoded.Init(1024)
+	offset := serialize.SUint32(100)
+	if err := encoded.Marshal(&offset); err != nil {
+		t.Fatalf("marshal offset: %v", err)
+	}
+	empty := &betweennessContributionReceiver{}
+	for index := 0; index < 10_000; index++ {
+		if err := encoded.Marshal(empty); err != nil {
+			t.Fatalf("marshal empty value %d: %v", index, err)
+		}
+	}
+	if got := len(encoded.PayLoad()); got != 4 {
+		t.Fatalf("10000 inactive vertices produced %d payload bytes, expected 4", got)
+	}
+}
+
+func TestBetweennessOutgoingChunksUseLocalVertexWindows(t *testing.T) {
+	worker := &BetweennessCentralityWorker{
+		localStart:     10,
+		localCount:     2,
+		outgoingChunks: make([]betweennessContributionChunk, 5),
+	}
+	context := worker.MakeContext(nil)
+	context.Step = 2
+	context.CreateValue(betweennessStopValue, compute.ValueTypeInt32,
+		compute.CValueActionAggregate)
+	context.SetValue(betweennessStopValue, serialize.SInt32(1))
+	worker.receiver.worker = worker
+
+	cases := []struct {
+		window int
+		offset int
+	}{
+		{window: 2, offset: 2},
+		{window: 2, offset: 4},
+		{window: 1, offset: 5},
+	}
+	for _, test := range cases {
+		if got := worker.outgoingWindowCount(); got != test.window {
+			t.Fatalf("offset %d expected window %d, got %d",
+				worker.outgoingOffset, test.window, got)
+		}
+		if _, ok := worker.VertexValue(10).(*betweennessContributionChunk); !ok {
+			t.Fatalf("offset %d did not expose a bounded contribution chunk",
+				worker.outgoingOffset)
+		}
+		worker.AfterStep()
+		if worker.outgoingOffset != test.offset {
+			t.Fatalf("expected outgoing offset %d, got %d",
+				test.offset, worker.outgoingOffset)
+		}
+		context.Step++
+	}
+	if stop := context.GetValue(betweennessStopValue).(serialize.SInt32); stop != 0 {
+		t.Fatalf("final chunk window did not stop: %d", stop)
+	}
+}
+
+func TestBetweennessMasterCompletesOnlyAfterScatterDrains(t *testing.T) {
+	master := &BetweennessCentralityMaster{}
+	context := master.MakeContext(map[string]string{"compute.max_step": "10"})
+	if err := master.Init(); err != nil {
+		t.Fatalf("master init: %v", err)
+	}
+	context.CreateValue(betweennessStopValue, compute.ValueTypeInt32,
+		compute.CValueActionAggregate)
+	context.SetValue(betweennessStopValue, serialize.SInt32(1))
+	if !master.Compute() {
+		t.Fatal("master stopped while a worker still had contribution chunks")
+	}
+	if complete := context.GetValue(betweennessCompleteValue).(serialize.SInt32); complete != 0 {
+		t.Fatalf("completion was published early: %d", complete)
+	}
+
+	context.SetValue(betweennessStopValue, serialize.SInt32(0))
+	if master.Compute() {
+		t.Fatal("master continued after every worker drained its chunks")
+	}
+	if complete := context.GetValue(betweennessCompleteValue).(serialize.SInt32); complete != 1 {
+		t.Fatalf("completion was not published: %d", complete)
+	}
+}
+
+func TestBetweennessOutputRequiresScalarCompletionAndUsesLocalRange(t *testing.T) {
+	worker := &BetweennessCentralityWorker{
+		localStart:  10,
+		localCount:  2,
+		localValues: []float64{1.5, 2.5},
+	}
+	context := worker.MakeContext(nil)
+	context.CreateValue(betweennessCompleteValue, compute.ValueTypeInt32,
+		compute.CValueActionNormal)
+	context.SetValue(betweennessCompleteValue, serialize.SInt32(0))
+	if err := worker.ValidateOutput(); err == nil {
+		t.Fatal("incomplete scalar state was accepted")
+	}
+
+	context.SetValue(betweennessCompleteValue, serialize.SInt32(1))
+	if err := worker.ValidateOutput(); err != nil {
+		t.Fatalf("completed local output rejected: %v", err)
+	}
+	context.Output = true
+	for vertexID, expected := range map[uint32]float32{10: 1.5, 11: 2.5} {
+		value, ok := worker.VertexValue(vertexID).(*serialize.SFloat32)
+		if !ok {
+			t.Fatalf("vertex %d output type is %T", vertexID, worker.VertexValue(vertexID))
+		}
+		if float32(*value) != expected {
+			t.Fatalf("vertex %d expected %v, got %v", vertexID, expected, *value)
+		}
+	}
+}