21180415886

Committed 20 Jan 2026 05:06PM UTC coverage: 70.81% (-0.1%) from 70.948%

Build # 21180415886

Build Type

push

github

Committed by

freeeve

Commit Message

fix(compaction): add SSTable reference counting to prevent use-after-close

Add reference counting to SSTables to prevent concurrent readers from
encountering "file already closed" errors during compaction.

Changes:
- Add refs and markedForRemoval fields to SSTable struct
- Add IncRef/DecRef/MarkForRemoval methods for safe lifecycle management
- Update reader.Get to hold refs while accessing SSTables
- Update ScanPrefix/ScanRange scanners to track and release refs
- Replace direct Close+Remove with MarkForRemoval in compaction

Fixes TestConcurrentReadsDuringCompaction race condition.

Run Details

50 of 53 new or added lines in 3 files covered. (94.34%)

760 existing lines in 12 files now uncovered.

5594 of 7900 relevant lines covered (70.81%)

405174.35 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

63.77

/batch_parallel.go

package tinykvs

import (
        "runtime"
        "sync"

        "github.com/freeeve/msgpck"
)

// KeyValue pairs a key with a struct value for bulk operations.
type KeyValue[T any] struct {
        Key   []byte
        Value *T
}

// PutStructs writes multiple structs in parallel using cached encoder.
// Parallelizes encoding across CPU cores, then writes atomically.
func PutStructs[T any](s *Store, items []KeyValue[T]) error {
        if len(items) == 0 {
                return nil
        }

        batch := NewBatch()
        if err := batchPutStructsParallel(batch, items, runtime.NumCPU()); err != nil {
                return err
        }

        return s.WriteBatch(batch)
}

// batchPutStructsParallel encodes structs in parallel using cached encoder.
func batchPutStructsParallel[T any](b *Batch, items []KeyValue[T], numWorkers int) error {
        if len(items) == 0 {
                return nil
        }

        enc := msgpck.GetStructEncoder[T]()

        if numWorkers <= 1 || len(items) < numWorkers {
                return encodeItemsSequential(b, enc, items)
        }

        ops, err := encodeItemsParallel(enc, items, numWorkers)
        if err != nil {
                return err
        }

        b.ops = append(b.ops, ops...)
        return nil
}

// encodeItemsSequential encodes items one at a time for small batches.
func encodeItemsSequential[T any](b *Batch, enc *msgpck.StructEncoder[T], items []KeyValue[T]) error {
        for _, item := range items {
                data, err := enc.EncodeCopy(item.Value)
                if err != nil {
                        return err
                }
                b.Put(item.Key, MsgpackValue(data))
        }
        return nil
}

// encodeItemsParallel encodes items across multiple workers.
func encodeItemsParallel[T any](enc *msgpck.StructEncoder[T], items []KeyValue[T], numWorkers int) ([]batchOp, error) {
        results := make([]batchOp, len(items))
        var firstErr error
        var errOnce sync.Once

        chunkSize := (len(items) + numWorkers - 1) / numWorkers
        var wg sync.WaitGroup

        for w := 0; w < numWorkers; w++ {
                start := w * chunkSize
                end := start + chunkSize
                if end > len(items) {
                        end = len(items)
                }
                if start >= end {
                        break
                }

                wg.Add(1)
                go func(start, end int) {
                        defer wg.Done()
                        encodeChunk(enc, items, results, start, end, &firstErr, &errOnce)
                }(start, end)
        }

        wg.Wait()

        if firstErr != nil {
                return nil, firstErr
        }

        return results, nil
}

// encodeChunk encodes a range of items into the results slice.
func encodeChunk[T any](enc *msgpck.StructEncoder[T], items []KeyValue[T], results []batchOp, start, end int, firstErr *error, errOnce *sync.Once) {
        for i := start; i < end; i++ {
                item := items[i]
                data, err := enc.EncodeCopy(item.Value)
                if err != nil {
                        errOnce.Do(func() { *firstErr = err })
                        return
                }
                keyCopy := make([]byte, len(item.Key))
                copy(keyCopy, item.Key)
                results[i] = batchOp{key: keyCopy, value: MsgpackValue(data), delete: false}
        }
}

1	package tinykvs
2
3	import (
4	"runtime"
5	"sync"
6
7	"github.com/freeeve/msgpck"
8	)
9
10	// KeyValue pairs a key with a struct value for bulk operations.
11	type KeyValue[T any] struct {
12	Key []byte
13	Value *T
14	}
15
16	// PutStructs writes multiple structs in parallel using cached encoder.
17	// Parallelizes encoding across CPU cores, then writes atomically.
18	func PutStructs[T any](s *Store, items []KeyValue[T]) error {	1✔
19	if len(items) == 0 {	1✔
UNCOV 20	return nil	×
UNCOV 21	}	×
22
23	batch := NewBatch()	1✔
24	if err := batchPutStructsParallel(batch, items, runtime.NumCPU()); err != nil {	1✔
25	return err	×
26	}	×
27
28	return s.WriteBatch(batch)	1✔
29	}
30
31	// batchPutStructsParallel encodes structs in parallel using cached encoder.
32	func batchPutStructsParallel[T any](b *Batch, items []KeyValue[T], numWorkers int) error {	1✔
33	if len(items) == 0 {	1✔
UNCOV 34	return nil	×
35	}	×
36
37	enc := msgpck.GetStructEncoder[T]()	1✔
38		1✔
39	if numWorkers <= 1 \|\| len(items) < numWorkers {	1✔
UNCOV 40	return encodeItemsSequential(b, enc, items)	×
UNCOV 41	}	×
42
43	ops, err := encodeItemsParallel(enc, items, numWorkers)	1✔
44	if err != nil {	1✔
UNCOV 45	return err	×
UNCOV 46	}	×
47
48	b.ops = append(b.ops, ops...)	1✔
49	return nil	1✔
50	}
51
52	// encodeItemsSequential encodes items one at a time for small batches.
UNCOV 53	func encodeItemsSequential[T any](b Batch, enc msgpck.StructEncoder[T], items []KeyValue[T]) error {	×
54	for _, item := range items {	×
UNCOV 55	data, err := enc.EncodeCopy(item.Value)	×
UNCOV 56	if err != nil {	×
UNCOV 57	return err	×
UNCOV 58	}	×
UNCOV 59	b.Put(item.Key, MsgpackValue(data))	×
60	}
UNCOV 61	return nil	×
62	}
63
64	// encodeItemsParallel encodes items across multiple workers.
65	func encodeItemsParallel[T any](enc *msgpck.StructEncoder[T], items []KeyValue[T], numWorkers int) ([]batchOp, error) {	1✔
66	results := make([]batchOp, len(items))	1✔
67	var firstErr error	1✔
68	var errOnce sync.Once	1✔
69		1✔
70	chunkSize := (len(items) + numWorkers - 1) / numWorkers	1✔
71	var wg sync.WaitGroup	1✔
72		1✔
73	for w := 0; w < numWorkers; w++ {	5✔
74	start := w * chunkSize	4✔
75	end := start + chunkSize	4✔
76	if end > len(items) {	4✔
77	end = len(items)	×
78	}	×
79	if start >= end {	4✔
UNCOV 80	break	×
81	}
82
83	wg.Add(1)	4✔
84	go func(start, end int) {	8✔
85	defer wg.Done()	4✔
86	encodeChunk(enc, items, results, start, end, &firstErr, &errOnce)	4✔
87	}(start, end)	4✔
88	}
89
90	wg.Wait()	1✔
91		1✔
92	if firstErr != nil {	1✔
UNCOV 93	return nil, firstErr	×
UNCOV 94	}	×
95
96	return results, nil	1✔
97	}
98
99	// encodeChunk encodes a range of items into the results slice.
100	func encodeChunk[T any](enc msgpck.StructEncoder[T], items []KeyValue[T], results []batchOp, start, end int, firstErr error, errOnce *sync.Once) {	4✔
101	for i := start; i < end; i++ {	1,004✔
102	item := items[i]	1,000✔
103	data, err := enc.EncodeCopy(item.Value)	1,000✔
104	if err != nil {	1,000✔
UNCOV 105	errOnce.Do(func() { *firstErr = err })	×
UNCOV 106	return	×
107	}
108	keyCopy := make([]byte, len(item.Key))	1,000✔
109	copy(keyCopy, item.Key)	1,000✔
110	results[i] = batchOp{key: keyCopy, value: MsgpackValue(data), delete: false}	1,000✔
111	}
112	}

freeeve / tinykvs / 21180415886

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous