diff --git a/api.go b/api.go index 96d2ec5..f766ba4 100644 --- a/api.go +++ b/api.go @@ -37,6 +37,9 @@ var SHA1ID = HashID([4]byte{'s', 'h', 'a', '1'}) var SHA256ID = HashID([4]byte{'s', '2', '5', '6'}) var NullHashID = HashID([4]byte{0, 0, 0, 0}) +// Size returns the hash size in bytes, or 0 if the ID is not one this +// package knows. Callers must treat 0 as an error: the hash ID can come +// straight out of a file header, so this must not panic. func (i HashID) Size() int { switch i { case NullHashID, SHA1ID: @@ -44,7 +47,7 @@ func (i HashID) Size() int { case SHA256ID: return 32 } - panic("unknown hash") + return 0 } // Table is a read interface for reftables, either file reftables or merged reftables. diff --git a/block.go b/block.go index cc26a07..6dc6b9c 100644 --- a/block.go +++ b/block.go @@ -175,9 +175,12 @@ func (br *blockReader) getType() byte { return br.block[br.headerOff] } -// newBlockWriter prepares for reading a block. +// newBlockReader prepares for reading a block. func newBlockReader(block []byte, headerOff uint32, tableBlockSize uint32, hashSize int) (*blockReader, error) { + if uint64(headerOff)+4 > uint64(len(block)) { + return nil, fmtError + } fullBlockSize := tableBlockSize typ := block[headerOff] if !isBlockType(typ) { @@ -185,8 +188,23 @@ func newBlockReader(block []byte, headerOff uint32, tableBlockSize uint32, hashS } sz := getU24(block[headerOff+1:]) + if uint64(sz) < uint64(headerOff)+6 { + return nil, fmtError + } + if typ != blockTypeLog && uint64(sz) > uint64(len(block)) { + return nil, fmtError + } if typ == blockTypeLog { + // sz is a 3-byte field read straight from the block header, and is + // deliberately not bounded by len(block) above because a log block + // declares its *decompressed* size. Bound it by what this input + // could possibly produce: DEFLATE cannot expand by more than + // 1032:1, so anything larger is malformed. Without this a ~40 byte + // table can reserve 16MiB up front, before a byte is decompressed. + if uint64(sz) > uint64(len(block))*maxDeflateRatio { + return nil, fmtError + } decompress := make([]byte, 0, sz) buf := bytes.NewBuffer(block) out := bytes.NewBuffer(decompress) @@ -194,19 +212,24 @@ func newBlockReader(block []byte, headerOff uint32, tableBlockSize uint32, hashS before := buf.Len() // Consume header - io.CopyN(out, buf, int64(headerOff+4)) + if _, err := io.CopyN(out, buf, int64(headerOff)+4); err != nil { + return nil, err + } r, err := zlib.NewReader(buf) if err != nil { return nil, err } - // Have to use io.Copy. zlib stream has a terminator, - // which we must consume, so go until EOF. - if _, err := io.Copy(out, r); err != nil { + defer r.Close() + // Read one byte beyond the declared payload size so an oversized + // stream is detected by the out.Len() != sz check below rather than + // being decompressed in full. Valid streams reach EOF within the + // limit and consume the zlib trailer, so the compressed-block + // accounting below (before - buf.Len()) stays correct. + limit := int64(sz) - int64(headerOff) - 4 + 1 + if _, err := io.Copy(out, io.LimitReader(r, limit)); err != nil { return nil, err } - r.Close() - if out.Len() != int(sz) { return nil, fmtError } @@ -228,7 +251,18 @@ func newBlockReader(block []byte, headerOff uint32, tableBlockSize uint32, hashS restartCount := binary.BigEndian.Uint16(block[len(block)-2:]) restartStart := len(block) - 2 - 3*int(restartCount) + if restartStart < int(headerOff)+4 { + return nil, fmtError + } restartBytes := block[restartStart:] + var previous uint32 + for i := 0; i < int(restartCount); i++ { + off := getU24(restartBytes[3*i:]) + if off < headerOff+4 || off >= uint32(restartStart) || (i > 0 && off <= previous) { + return nil, fmtError + } + previous = off + } block = block[:restartStart] br := &blockReader{ diff --git a/block_validation_test.go b/block_validation_test.go new file mode 100644 index 0000000..d02ad28 --- /dev/null +++ b/block_validation_test.go @@ -0,0 +1,43 @@ +package reftable + +import ( + "bytes" + "compress/zlib" + "testing" +) + +func TestBlockReaderRejectsInvalidLayout(t *testing.T) { + for _, tc := range []struct { + name string + data []byte + headerOff uint32 + }{ + {"missing_header", nil, 0}, + {"short_header", []byte{'r', 0, 0}, 0}, + {"header_offset", []byte{'r', 0, 0, 6, 0, 0}, 24}, + {"oversized_restart_count", []byte{'r', 0, 0, 6, 255, 255}, 0}, + {"restart_in_header", []byte{'r', 0, 0, 10, 0, 0, 0, 0, 0, 1}, 0}, + {"restart_in_trailer", []byte{'r', 0, 0, 10, 0, 0, 0, 5, 0, 1}, 0}, + } { + t.Run(tc.name, func(t *testing.T) { + if _, err := newBlockReader(tc.data, tc.headerOff, 0, 20); err == nil { + t.Fatal("invalid block layout accepted") + } + }) + } +} + +func TestBlockReaderRejectsOversizedLogStream(t *testing.T) { + var compressed bytes.Buffer + compressed.Write([]byte{'g', 0, 0, 6}) // Declares only two payload bytes. + zw := zlib.NewWriter(&compressed) + if _, err := zw.Write(bytes.Repeat([]byte{0}, 1<<20)); err != nil { + t.Fatal(err) + } + if err := zw.Close(); err != nil { + t.Fatal(err) + } + if _, err := newBlockReader(compressed.Bytes(), 0, 0, 20); err == nil { + t.Fatal("oversized decompressed payload accepted") + } +} diff --git a/c/block.c b/c/block.c index 855e3f5..cc43432 100644 --- a/c/block.c +++ b/c/block.c @@ -13,6 +13,7 @@ license that can be found in the LICENSE file or at #include "record.h" #include "reftable-error.h" #include "system.h" +#include #include int header_size(int version) @@ -186,13 +187,22 @@ int block_reader_init(struct block_reader *br, struct reftable_block *block, int hash_size) { uint32_t full_block_size = table_block_size; - uint8_t typ = block->data[header_off]; - uint32_t sz = get_be24(block->data + header_off + 1); + uint8_t typ; + uint32_t sz; uint16_t restart_count = 0; uint32_t restart_start = 0; uint8_t *restart_bytes = NULL; + uint32_t previous = 0; + int i; + if ((uint64_t)header_off + 4 > block->len) + return REFTABLE_FORMAT_ERROR; + typ = block->data[header_off]; + sz = get_be24(block->data + header_off + 1); + if ((uint64_t)sz < (uint64_t)header_off + 6 || + (typ != BLOCK_TYPE_LOG && sz > block->len)) + return REFTABLE_FORMAT_ERROR; if (!reftable_is_block_type(typ)) return REFTABLE_FORMAT_ERROR; @@ -216,8 +226,10 @@ int block_reader_init(struct block_reader *br, struct reftable_block *block, return REFTABLE_ZLIB_ERROR; } - if (dst_len + block_header_skip != sz) + if (dst_len + block_header_skip != sz) { + reftable_free(uncompressed); return REFTABLE_FORMAT_ERROR; + } /* We're done with the input data. */ reftable_block_done(block); @@ -236,8 +248,17 @@ int block_reader_init(struct block_reader *br, struct reftable_block *block, } restart_count = get_be16(block->data + sz - 2); + if (2 + 3 * (uint32_t)restart_count > sz - header_off - 4) + return REFTABLE_FORMAT_ERROR; restart_start = sz - 2 - 3 * restart_count; restart_bytes = block->data + restart_start; + for (i = 0; i < restart_count; i++) { + uint32_t off = get_be24(restart_bytes + 3 * i); + if (off < header_off + 4 || off >= restart_start || + (i > 0 && off <= previous)) + return REFTABLE_FORMAT_ERROR; + previous = off; + } /* transfer ownership. */ br->block = *block; diff --git a/c/block_test.c b/c/block_test.c index 4b3ea26..cdd52d4 100644 --- a/c/block_test.c +++ b/c/block_test.c @@ -113,8 +113,38 @@ static void test_block_read_write(void) } } +static void test_block_rejects_invalid_layout(void) +{ + uint8_t data[][10] = { + { 'r', 0, 0, 0 }, + { 'r', 0, 0, 1 }, + { 'r', 255, 255, 255 }, + { 'r', 0, 0, 6, 255, 255 }, + { 'r', 0, 0, 10, 0, 0, 0, 0, 0, 1 }, + { 'r', 0, 0, 10, 0, 0, 0, 5, 0, 1 }, + { 'g', 0, 0, 1 }, + }; + int i; + for (i = 0; i < ARRAY_SIZE(data); i++) { + struct block_reader br = { 0 }; + struct reftable_block block = { + .data = data[i], + .len = sizeof(data[i]), + }; + int err = block_reader_init(&br, &block, 0, 0, GIT_SHA1_RAWSZ); + EXPECT(err == REFTABLE_FORMAT_ERROR); + } + for (i = 0; i < 4; i++) { + struct block_reader br = { 0 }; + struct reftable_block block = { .data = data[0], .len = i }; + int err = block_reader_init(&br, &block, 0, 0, GIT_SHA1_RAWSZ); + EXPECT(err == REFTABLE_FORMAT_ERROR); + } +} + int block_test_main(int argc, const char *argv[]) { RUN_TEST(test_block_read_write); + RUN_TEST(test_block_rejects_invalid_layout); return 0; } diff --git a/c/iter.c b/c/iter.c index 93d04f7..3a8fc35 100644 --- a/c/iter.c +++ b/c/iter.c @@ -146,12 +146,15 @@ static int indexed_table_ref_iter_next(void *p, struct reftable_record *rec) } continue; } - /* BUG */ - if (!memcmp(it->oid.buf, ref->value.val2.target_value, - it->oid.len) || - !memcmp(it->oid.buf, ref->value.val2.value, it->oid.len)) { + ref->update_index += it->r->min_update_index; + if (ref->value_type == REFTABLE_REF_VAL2 && + (!memcmp(it->oid.buf, ref->value.val2.target_value, + it->oid.len) || + !memcmp(it->oid.buf, ref->value.val2.value, it->oid.len))) + return 0; + if (ref->value_type == REFTABLE_REF_VAL1 && + !memcmp(it->oid.buf, ref->value.val1, it->oid.len)) return 0; - } } } diff --git a/c/reader.c b/c/reader.c index ab7a981..1bf1ec6 100644 --- a/c/reader.c +++ b/c/reader.c @@ -292,10 +292,16 @@ int reader_init_block_reader(struct reftable_reader *r, struct block_reader *br, if (err < 0) return err; + if ((uint64_t)header_off + 4 > block.len) { + reftable_block_done(&block); + return REFTABLE_FORMAT_ERROR; + } block_size = extract_block_size(block.data, &block_typ, next_off, r->version); - if (block_size < 0) - return block_size; + if (!reftable_is_block_type(block_typ)) { + reftable_block_done(&block); + return REFTABLE_FORMAT_ERROR; + } if (want_typ != BLOCK_TYPE_ANY && block_typ != want_typ) { reftable_block_done(&block); @@ -310,8 +316,11 @@ int reader_init_block_reader(struct reftable_reader *r, struct block_reader *br, } } - return block_reader_init(br, &block, header_off, r->block_size, + err = block_reader_init(br, &block, header_off, r->block_size, hash_size(r->hash_id)); + if (err < 0) + reftable_block_done(&block); + return err; } static int table_iter_next_block(struct table_iter *dest, diff --git a/c/readwrite_test.c b/c/readwrite_test.c index 5f6bcc2..f2a1926 100644 --- a/c/readwrite_test.c +++ b/c/readwrite_test.c @@ -421,7 +421,7 @@ static void test_table_read_write_seek_index(void) test_table_read_write_seek(1, GIT_SHA1_FORMAT_ID); } -static void test_table_refs_for(int indexed) +static void test_table_refs_for(int indexed, int peeled) { int N = 50; char **want_names = reftable_calloc(sizeof(char *) * (N + 1)); @@ -446,6 +446,7 @@ static void test_table_refs_for(int indexed) int j; set_test_hash(want_hash, 4); + reftable_writer_set_limits(w, 100, 200); for (i = 0; i < N; i++) { uint8_t hash[GIT_SHA1_RAWSZ]; @@ -464,9 +465,15 @@ static void test_table_refs_for(int indexed) set_test_hash(hash1, i / 4); set_test_hash(hash2, 3 + i / 4); - ref.value_type = REFTABLE_REF_VAL2; - ref.value.val2.value = hash1; - ref.value.val2.target_value = hash2; + ref.update_index = 150; + if (peeled) { + ref.value_type = REFTABLE_REF_VAL2; + ref.value.val2.value = hash1; + ref.value.val2.target_value = hash2; + } else { + ref.value_type = REFTABLE_REF_VAL1; + ref.value.val1 = hash1; + } /* 80 bytes / entry, so 3 entries per block. Yields 17 */ @@ -475,7 +482,7 @@ static void test_table_refs_for(int indexed) EXPECT(n == 0); if (!memcmp(hash1, want_hash, GIT_SHA1_RAWSZ) || - !memcmp(hash2, want_hash, GIT_SHA1_RAWSZ)) { + (peeled && !memcmp(hash2, want_hash, GIT_SHA1_RAWSZ))) { want_names[want_names_len++] = xstrdup(name); } } @@ -511,6 +518,7 @@ static void test_table_refs_for(int indexed) EXPECT(j < want_names_len); EXPECT(0 == strcmp(ref.refname, want_names[j])); + EXPECT(ref.update_index == 150); j++; reftable_ref_record_release(&ref); } @@ -524,12 +532,14 @@ static void test_table_refs_for(int indexed) static void test_table_refs_for_no_index(void) { - test_table_refs_for(0); + test_table_refs_for(0, 0); + test_table_refs_for(0, 1); } static void test_table_refs_for_obj_index(void) { - test_table_refs_for(1); + test_table_refs_for(1, 0); + test_table_refs_for(1, 1); } static void test_write_empty_table(void) diff --git a/c/stack.c b/c/stack.c index df5021e..8e1100a 100644 --- a/c/stack.c +++ b/c/stack.c @@ -198,6 +198,8 @@ static int reftable_stack_reload_once(struct reftable_stack *st, char **names, struct reftable_reader **cur = stack_copy_readers(st, cur_len); int err = 0; int names_len = names_length(names); + char **retained_names = names; + int *borrowed = reftable_calloc(sizeof(int) * names_len); struct reftable_reader **new_readers = reftable_calloc(sizeof(struct reftable_reader *) * names_len); struct reftable_table *new_tables = @@ -216,6 +218,7 @@ static int reftable_stack_reload_once(struct reftable_stack *st, char **names, for (j = 0; reuse_open && j < cur_len; j++) { if (cur[j] && 0 == strcmp(cur[j]->name, name)) { rd = cur[j]; + borrowed[new_readers_len] = 1; cur[j] = NULL; break; } @@ -268,13 +271,15 @@ static int reftable_stack_reload_once(struct reftable_stack *st, char **names, if (cur[i]) { const char *name = reader_name(cur[i]); struct strbuf filename = STRBUF_INIT; - stack_filename(&filename, st, name); + if (!has_name(retained_names, name)) + stack_filename(&filename, st, name); reader_close(cur[i]); reftable_reader_free(cur[i]); /* On Windows, can only unlink after closing. */ - unlink(filename.buf); + if (filename.len) + unlink(filename.buf); strbuf_release(&filename); } @@ -282,9 +287,10 @@ static int reftable_stack_reload_once(struct reftable_stack *st, char **names, done: for (i = 0; i < new_readers_len; i++) { - reader_close(new_readers[i]); - reftable_reader_free(new_readers[i]); + if (!borrowed[i]) + reftable_reader_free(new_readers[i]); } + reftable_free(borrowed); reftable_free(new_readers); reftable_free(new_tables); reftable_free(cur); @@ -445,6 +451,7 @@ static void format_name(struct strbuf *dest, uint64_t min, uint64_t max) struct reftable_addition { int lock_file_fd; + int have_lock; struct strbuf lock_file_name; struct reftable_stack *stack; @@ -455,6 +462,7 @@ struct reftable_addition { #define REFTABLE_ADDITION_INIT \ { \ + .lock_file_fd = -1, \ .lock_file_name = STRBUF_INIT \ } @@ -478,11 +486,12 @@ static int reftable_stack_init_addition(struct reftable_addition *add, } goto done; } + add->have_lock = 1; err = stack_uptodate(st); if (err < 0) goto done; - if (err > 1) { + if (err > 0) { err = REFTABLE_LOCK_ERROR; goto done; } @@ -509,14 +518,15 @@ static void reftable_addition_close(struct reftable_addition *add) add->new_tables = NULL; add->new_tables_len = 0; - if (add->lock_file_fd > 0) { + if (add->lock_file_fd >= 0) { close(add->lock_file_fd); - add->lock_file_fd = 0; + add->lock_file_fd = -1; } - if (add->lock_file_name.len > 0) { + if (add->have_lock) { unlink(add->lock_file_name.buf); - strbuf_release(&add->lock_file_name); + add->have_lock = 0; } + strbuf_release(&add->lock_file_name); strbuf_release(&nm); } @@ -555,7 +565,7 @@ int reftable_addition_commit(struct reftable_addition *add) } err = close(add->lock_file_fd); - add->lock_file_fd = 0; + add->lock_file_fd = -1; if (err < 0) { err = REFTABLE_IO_ERROR; goto done; @@ -568,6 +578,7 @@ int reftable_addition_commit(struct reftable_addition *add) } /* success, no more state to clean up. */ + add->have_lock = 0; strbuf_release(&add->lock_file_name); for (i = 0; i < add->new_tables_len; i++) { reftable_free(add->new_tables[i]); @@ -818,6 +829,10 @@ static int stack_write_compact(struct reftable_stack *st, } entries++; } + /* Do not overwrite a ref read/write error with a successful log seek: + * publishing the partial output would permanently drop unread refs. */ + if (err < 0) + goto done; reftable_iterator_destroy(&it); err = reftable_merged_table_seek_log(mt, &it, ""); @@ -880,6 +895,8 @@ static int stack_compact_range(struct reftable_stack *st, int first, int last, int lock_file_fd = 0; int compact_count = last - first + 1; char **listp = NULL; + char **current_names = NULL; + int current_len = 0, replacement_start = -1; char **delete_on_success = reftable_calloc(sizeof(char *) * (compact_count + 1)); char **subtable_locks = @@ -932,15 +949,13 @@ static int stack_compact_range(struct reftable_stack *st, int first, int last, sublock_file_fd = open(subtab_lock.buf, O_EXCL | O_CREAT | O_WRONLY, 0644); - if (sublock_file_fd > 0) { - close(sublock_file_fd); - } else if (sublock_file_fd < 0) { - if (errno == EEXIST) { - err = 1; - } else { - err = REFTABLE_IO_ERROR; - } + if (sublock_file_fd < 0) { + err = errno == EEXIST ? 1 : REFTABLE_IO_ERROR; + strbuf_release(&subtab_lock); + strbuf_release(&subtab_file_name); + goto done; } + close(sublock_file_fd); subtable_locks[j] = subtab_lock.buf; delete_on_success[j] = subtab_file_name.buf; @@ -978,6 +993,27 @@ static int stack_compact_range(struct reftable_stack *st, int first, int last, } have_lock = 1; + /* Preserve changes committed while we did not hold the global lock. */ + err = read_lines(st->list_file, ¤t_names); + if (err < 0) + goto done; + current_len = names_length(current_names); + for (i = 0; i + compact_count <= current_len; i++) { + for (j = 0; j < compact_count; j++) { + if (strcmp(current_names[i + j], + st->readers[first + j]->name)) + break; + } + if (j == compact_count) { + replacement_start = i; + break; + } + } + if (replacement_start < 0) { + err = 1; + goto done; + } + format_name(&new_table_name, st->readers[first]->min_update_index, st->readers[last]->max_update_index); strbuf_addstr(&new_table_name, ".ref"); @@ -993,16 +1029,16 @@ static int stack_compact_range(struct reftable_stack *st, int first, int last, } } - for (i = 0; i < first; i++) { - strbuf_addstr(&ref_list_contents, st->readers[i]->name); + for (i = 0; i < replacement_start; i++) { + strbuf_addstr(&ref_list_contents, current_names[i]); strbuf_addstr(&ref_list_contents, "\n"); } if (!is_empty_table) { strbuf_addbuf(&ref_list_contents, &new_table_name); strbuf_addstr(&ref_list_contents, "\n"); } - for (i = last + 1; i < st->merged->stack_len; i++) { - strbuf_addstr(&ref_list_contents, st->readers[i]->name); + for (i = replacement_start + compact_count; i < current_len; i++) { + strbuf_addstr(&ref_list_contents, current_names[i]); strbuf_addstr(&ref_list_contents, "\n"); } @@ -1042,6 +1078,9 @@ static int stack_compact_range(struct reftable_stack *st, int first, int last, } done: + free_names(current_names); + if (temp_tab_file_name.len) + unlink(temp_tab_file_name.buf); free_names(delete_on_success); listp = subtable_locks; diff --git a/c/stack_test.c b/c/stack_test.c index 7917632..f96ee44 100644 --- a/c/stack_test.c +++ b/c/stack_test.c @@ -11,6 +11,7 @@ license that can be found in the LICENSE file or at #include "system.h" #include "reftable-reader.h" +#include "reader.h" #include "merged.h" #include "basics.h" #include "constants.h" @@ -358,6 +359,265 @@ static void test_reftable_stack_lock_failure(void) clear_dir(dir); } +/* Commit through a second stack on the first compaction read, after the + * compactor releases tables.list.lock. No sleeps or process scheduling needed. */ +struct addition_on_read { + struct reftable_block_source original; + struct reftable_stack *writer; + struct reftable_ref_record ref; + int fired; +}; + +static uint64_t addition_on_read_size(void *arg) +{ + struct addition_on_read *hook = arg; + return block_source_size(&hook->original); +} + +static int addition_on_read_block(void *arg, struct reftable_block *dest, + uint64_t off, uint32_t size) +{ + struct addition_on_read *hook = arg; + if (!hook->fired) { + int err; + hook->fired = 1; + err = reftable_stack_add(hook->writer, &write_test_ref, &hook->ref); + EXPECT_ERR(err); + } + return block_source_read_block(&hook->original, dest, off, size); +} + +static void addition_on_read_return(void *arg, struct reftable_block *block) +{ + struct addition_on_read *hook = arg; + hook->original.ops->return_block(hook->original.arg, block); +} + +static void addition_on_read_close(void *arg) +{ + struct addition_on_read *hook = arg; + block_source_close(&hook->original); +} + +static struct reftable_block_source_vtable addition_on_read_ops = { + .size = addition_on_read_size, + .read_block = addition_on_read_block, + .return_block = addition_on_read_return, + .close = addition_on_read_close, +}; + +static void test_reftable_stack_compaction_preserves_addition(void) +{ + char *dir = get_tmp_dir(__LINE__); + struct reftable_write_options cfg = { 0 }; + struct reftable_stack *compactor = NULL, *writer = NULL, *fresh = NULL; + struct reftable_ref_record ref = { + .refname = "refs/heads/a", + .update_index = 1, + .value_type = REFTABLE_REF_SYMREF, + .value.symref = "refs/heads/main", + }; + struct reftable_ref_record got = { NULL }; + struct addition_on_read hook = { 0 }; + int err; + + err = reftable_new_stack(&compactor, dir, cfg); + EXPECT_ERR(err); + compactor->disable_auto_compact = 1; + err = reftable_stack_add(compactor, &write_test_ref, &ref); + EXPECT_ERR(err); + ref.refname = "refs/heads/b"; + ref.update_index = 2; + err = reftable_stack_add(compactor, &write_test_ref, &ref); + EXPECT_ERR(err); + err = reftable_new_stack(&writer, dir, cfg); + EXPECT_ERR(err); + writer->disable_auto_compact = 1; + ref.refname = "refs/heads/c"; + ref.update_index = 3; + hook.original = compactor->readers[0]->source; + hook.writer = writer; + hook.ref = ref; + compactor->readers[0]->source.ops = &addition_on_read_ops; + compactor->readers[0]->source.arg = &hook; + + err = reftable_stack_compact_all(compactor, NULL); + EXPECT_ERR(err); + EXPECT(hook.fired); + err = reftable_new_stack(&fresh, dir, cfg); + EXPECT_ERR(err); + err = reftable_stack_read_ref(fresh, "refs/heads/c", &got); + EXPECT_ERR(err); + EXPECT(got.update_index == 3); + reftable_ref_record_release(&got); + reftable_stack_destroy(fresh); + reftable_stack_destroy(writer); + reftable_stack_destroy(compactor); + clear_dir(dir); +} + +static int compaction_read_failure(void *arg, struct reftable_block *dest, + uint64_t off, uint32_t size) +{ + struct addition_on_read *hook = arg; + /* Let seek and initial iteration succeed, then fail at the third block. */ + if (off >= 256) { + hook->fired = 1; + return REFTABLE_IO_ERROR; + } + return block_source_read_block(&hook->original, dest, off, size); +} + +static int write_compaction_refs(struct reftable_writer *wr, void *arg) +{ + int count = *(int *)arg; + uint8_t oid[GIT_SHA1_RAWSZ] = { 1 }; + int i; + reftable_writer_set_limits(wr, 1, 1); + for (i = 0; i < count; i++) { + char name[64]; + struct reftable_ref_record ref = { NULL }; + int err; + snprintf(name, sizeof(name), "refs/heads/branch%04d", i); + ref.refname = name; + ref.update_index = 1; + ref.value_type = REFTABLE_REF_VAL1; + ref.value.val1 = oid; + err = reftable_writer_add_ref(wr, &ref); + if (err) + return err; + } + return 0; +} + +static void test_reftable_stack_compaction_preserves_read_error(void) +{ + char *dir = get_tmp_dir(__LINE__); + struct reftable_write_options cfg = { .block_size = 128 }; + struct reftable_stack *st = NULL, *fresh = NULL; + struct reftable_log_expiry_config expiry = { 0 }; + struct addition_on_read hook = { 0 }; + struct reftable_block_source_vtable ops = addition_on_read_ops; + char **before = NULL, **after = NULL; + int count = 9, err, i; + + err = reftable_new_stack(&st, dir, cfg); + EXPECT_ERR(err); + st->disable_auto_compact = 1; + err = reftable_stack_add(st, &write_compaction_refs, &count); + EXPECT_ERR(err); + /* Three data blocks without an index: the error occurs during iteration, + * not during the initial seek that already propagates errors correctly. */ + EXPECT(st->readers[0]->ref_offsets.index_offset == 0); + EXPECT(st->readers[0]->size > 256); + err = read_lines(st->list_file, &before); + EXPECT_ERR(err); + hook.original = st->readers[0]->source; + ops.read_block = compaction_read_failure; + st->readers[0]->source.ops = &ops; + st->readers[0]->source.arg = &hook; + + /* Non-NULL expiry forces compaction even for a single table. */ + err = reftable_stack_compact_all(st, &expiry); + EXPECT(hook.fired); + EXPECT(err == REFTABLE_IO_ERROR); + st->readers[0]->source = hook.original; + err = read_lines(st->list_file, &after); + EXPECT_ERR(err); + EXPECT(names_equal(before, after)); + EXPECT(count_dir_entries(dir) == 2); /* Original table and manifest only. */ + + err = reftable_new_stack(&fresh, dir, cfg); + EXPECT_ERR(err); + for (i = 0; i < count; i++) { + char name[64]; + struct reftable_ref_record ref = { NULL }; + snprintf(name, sizeof(name), "refs/heads/branch%04d", i); + err = reftable_stack_read_ref(st, name, &ref); + EXPECT_ERR(err); + reftable_ref_record_release(&ref); + err = reftable_stack_read_ref(fresh, name, &ref); + EXPECT_ERR(err); + EXPECT(ref.update_index == 1); + reftable_ref_record_release(&ref); + } + free_names(before); + free_names(after); + reftable_stack_destroy(fresh); + reftable_stack_destroy(st); + clear_dir(dir); +} + +static void test_reftable_stack_reload_failure_preserves_readers(void) +{ + char *dir = get_tmp_dir(__LINE__); + struct reftable_write_options cfg = { 0 }; + struct reftable_stack *st = NULL; + struct reftable_ref_record ref = { + .refname = "HEAD", + .update_index = 1, + .value_type = REFTABLE_REF_SYMREF, + .value.symref = "refs/heads/main", + }; + struct reftable_ref_record got = { NULL }; + const char missing[] = "missing.ref\n"; + int err, fd; + + err = reftable_new_stack(&st, dir, cfg); + EXPECT_ERR(err); + err = reftable_stack_add(st, &write_test_ref, &ref); + EXPECT_ERR(err); + fd = open(st->list_file, O_WRONLY | O_APPEND); + EXPECT(fd >= 0); + EXPECT(write(fd, missing, sizeof(missing) - 1) == sizeof(missing) - 1); + EXPECT(close(fd) == 0); + + err = reftable_stack_reload(st); + EXPECT(err == REFTABLE_NOT_EXIST_ERROR); + err = reftable_stack_read_ref(st, "HEAD", &got); + EXPECT_ERR(err); + EXPECT_STREQ(got.value.symref, "refs/heads/main"); + reftable_ref_record_release(&got); + reftable_stack_destroy(st); + clear_dir(dir); +} + +/* A contender that never owned the lock must not unlink it on failure. */ +static void test_reftable_stack_failed_addition_preserves_lock(void) +{ + char *dir = get_tmp_dir(__LINE__); + struct reftable_write_options cfg = { 0 }; + struct reftable_stack *owner = NULL; + struct reftable_stack *contender = NULL; + struct reftable_addition *first = NULL; + struct reftable_addition *second = NULL; + struct strbuf lock = STRBUF_INIT; + int err, contender_err, lock_survived; + + err = reftable_new_stack(&owner, dir, cfg); + EXPECT_ERR(err); + err = reftable_new_stack(&contender, dir, cfg); + EXPECT_ERR(err); + err = reftable_stack_new_addition(&first, owner); + EXPECT_ERR(err); + strbuf_addstr(&lock, dir); + strbuf_addstr(&lock, "/tables.list.lock"); + EXPECT(access(lock.buf, F_OK) == 0); + + contender_err = reftable_stack_new_addition(&second, contender); + lock_survived = access(lock.buf, F_OK) == 0; + + reftable_addition_destroy(second); + reftable_addition_destroy(first); + reftable_stack_destroy(contender); + reftable_stack_destroy(owner); + strbuf_release(&lock); + clear_dir(dir); + + EXPECT(contender_err == REFTABLE_LOCK_ERROR); + EXPECT(lock_survived); +} + static void test_reftable_stack_add(void) { int i = 0; @@ -936,6 +1196,10 @@ int stack_test_main(int argc, const char *argv[]) RUN_TEST(test_reftable_stack_compaction_concurrent_clean); RUN_TEST(test_reftable_stack_hash_id); RUN_TEST(test_reftable_stack_lock_failure); + RUN_TEST(test_reftable_stack_compaction_preserves_addition); + RUN_TEST(test_reftable_stack_compaction_preserves_read_error); + RUN_TEST(test_reftable_stack_failed_addition_preserves_lock); + RUN_TEST(test_reftable_stack_reload_failure_preserves_readers); RUN_TEST(test_reftable_stack_log_normalize); RUN_TEST(test_reftable_stack_tombstone); RUN_TEST(test_reftable_stack_transaction_api); diff --git a/c/strbuf.h b/c/strbuf.h index 2f1836d..9199f14 100644 --- a/c/strbuf.h +++ b/c/strbuf.h @@ -11,6 +11,7 @@ license that can be found in the LICENSE file or at #include #include +#include /* * Provides a bounds-checked, growable byte ranges. To use, initialize as diff --git a/constants.go b/constants.go index 2eb51d3..4627eed 100644 --- a/constants.go +++ b/constants.go @@ -35,3 +35,14 @@ const blockTypeObj = 'o' const blockTypeAny = 0 const maxRestarts = (1 << 16) - 1 + +// maxDeflateRatio is the theoretical maximum expansion of a DEFLATE stream. +// Used to reject log blocks whose declared decompressed size cannot possibly +// be produced by the compressed bytes actually present. +const maxDeflateRatio = 1032 + +// maxIndexDepth bounds how many index levels a seek will descend. Offsets in +// index records are file-supplied and can form a cycle; a valid tree is only +// a few levels deep because each level holds strictly fewer blocks than the +// one below it. +const maxIndexDepth = 64 diff --git a/fuzz_test.go b/fuzz_test.go new file mode 100644 index 0000000..c702d36 --- /dev/null +++ b/fuzz_test.go @@ -0,0 +1,161 @@ +package reftable + +import ( + "bytes" + "fmt" + "testing" +) + +// corpusTable writes a table with enough refs and reflogs to build a +// multi-level index for every section, which is where the offsets that a +// reader follows out of the file live. +func corpusTable(t testing.TB, refs, logs int, cfg Config) []byte { + t.Helper() + + hashSize := 20 + if cfg.HashID == SHA256ID { + hashSize = 32 + } + hash := func(i int) []byte { + h := make([]byte, hashSize) + h[hashSize-1], h[hashSize-2] = byte(i), byte(i>>8) + return h + } + + buf := &bytes.Buffer{} + w, err := NewWriter(buf, &cfg) + if err != nil { + t.Fatal(err) + } + w.SetLimits(1, 1) + for i := range refs { + if err := w.AddRef(&RefRecord{ + RefName: fmt.Sprintf("refs/heads/b%06d", i), UpdateIndex: 1, Value: hash(i), + }); err != nil { + t.Fatal(err) + } + } + for i := range logs { + if err := w.AddLog(&LogRecord{ + RefName: fmt.Sprintf("refs/heads/b%06d", i), UpdateIndex: 1, + New: hash(i), Old: hash(i), Name: "n", Email: "e@x", Message: "m", + }); err != nil { + t.Fatal(err) + } + } + if err := w.Close(); err != nil { + t.Fatal(err) + } + return buf.Bytes() +} + +// driveReader exercises every path that follows an offset, length, or block +// type read out of the table. It must never panic, whatever the bytes say. +func driveReader(data []byte) { + rd, err := NewReader(&ByteBlockSource{Source: data}, "fuzz") + if err != nil { + return + } + _ = rd.MaxUpdateIndex() + _ = rd.MinUpdateIndex() + + for _, key := range []string{"", "refs/heads/b000000", "refs/heads/b000250", "refs/heads/zzz"} { + if it, err := rd.SeekRef(key); err == nil && it != nil { + for range 4 { + var r RefRecord + if ok, err := it.NextRef(&r); !ok || err != nil { + break + } + } + } + if it, err := rd.SeekLog(key, 1); err == nil && it != nil { + for range 4 { + var l LogRecord + if ok, err := it.NextLog(&l); !ok || err != nil { + break + } + } + } + } + + // Both hash sizes: objectIDLen comes from the footer and need not match. + for _, n := range []int{20, 32} { + oid := make([]byte, n) + oid[n-1] = 0xfa + if it, err := rd.RefsFor(oid); err == nil && it != nil { + var r RefRecord + it.NextRef(&r) + } + } +} + +// A corrupt table must produce an error, never a panic. Every byte of a +// reftable is attacker-controlled: it can arrive from a clone, a hostile +// archive, or a partially written file after a crash. +// +// This is a deterministic sweep so it runs in CI; FuzzReader below explores +// the same surface without a fixed stride. +func TestReaderSurvivesCorruption(t *testing.T) { + shapes := []struct { + name string + refs, logs int + cfg Config + }{ + {"sha1/bs256", 500, 500, Config{BlockSize: 256}}, + {"sha1/unaligned", 300, 300, Config{BlockSize: 256, Unaligned: true}}, + {"sha256/bs256", 100, 100, Config{BlockSize: 256, HashID: SHA256ID}}, + } + + for _, s := range shapes { + t.Run(s.name, func(t *testing.T) { + base := corpusTable(t, s.refs, s.logs, s.cfg) + + // Stride keeps this fast while still covering every region: + // headers, block trailers, restart tables and index offsets. + const stride = 3 + for pos := 0; pos < len(base); pos += stride { + for _, v := range []byte{0x00, 0x80, 0xff} { + if base[pos] == v { + continue + } + mutated := append([]byte(nil), base...) + mutated[pos] = v + + func() { + defer func() { + if r := recover(); r != nil { + t.Fatalf("panic on byte %d = %#02x: %v", pos, v, r) + } + }() + driveReader(mutated) + }() + } + } + + // Truncation at every length must also be handled. + for n := 0; n < len(base); n += 17 { + func() { + defer func() { + if r := recover(); r != nil { + t.Fatalf("panic on truncation to %d bytes: %v", n, r) + } + }() + driveReader(base[:n]) + }() + } + }) + } +} + +func FuzzReader(f *testing.F) { + f.Add(corpusTable(f, 500, 500, Config{BlockSize: 256})) + f.Add(corpusTable(f, 300, 300, Config{BlockSize: 256, Unaligned: true})) + f.Add(corpusTable(f, 100, 100, Config{BlockSize: 256, HashID: SHA256ID})) + f.Add(corpusTable(f, 60, 60, Config{})) + f.Add([]byte{}) + f.Add(make([]byte, 1024)) + + f.Fuzz(func(t *testing.T, data []byte) { + driveReader(data) + }) +} diff --git a/merged.go b/merged.go index 3763b2c..f04c303 100644 --- a/merged.go +++ b/merged.go @@ -127,11 +127,17 @@ func NewMerged(tabs []Table, hashID [4]byte) (*Merged, error) { // MaxUpdateIndex implements the Table interface. func (m *Merged) MaxUpdateIndex() uint64 { + if len(m.stack) == 0 { + return 0 + } return m.stack[len(m.stack)-1].MaxUpdateIndex() } // MinUpdateIndex implements the Table interface. func (m *Merged) MinUpdateIndex() uint64 { + if len(m.stack) == 0 { + return 0 + } return m.stack[0].MinUpdateIndex() } @@ -198,7 +204,7 @@ func (m *Merged) seekRecord(rec record) (iterator, error) { for _, t := range m.stack { iter, err := t.seekRecord(rec) if err != nil { - return nil, fmt.Errorf("reftable: seek %s: %v", t.Name(), err) + return nil, fmt.Errorf("reftable: seek %s: %w", t.Name(), err) } its = append(its, iter) names = append(names, t.Name()) @@ -237,7 +243,7 @@ func (it *mergedIter) init() error { rec := newRecord(it.typ, "") ok, err := sub.Next(rec) if err != nil { - return fmt.Errorf("init %s: %v", it.names[i], err) + return fmt.Errorf("init %s: %w", it.names[i], err) } if ok { it.pq.add(pqEntry{ @@ -261,7 +267,7 @@ func (m *mergedIter) advanceSubIter(index int) error { r := newRecord(m.typ, "") ok, err := m.stack[index].Next(r) if err != nil { - return fmt.Errorf("next %s: %v", m.names[index], err) + return fmt.Errorf("next %s: %w", m.names[index], err) } if !ok { diff --git a/reader.go b/reader.go index a7ec313..9ac982e 100644 --- a/reader.go +++ b/reader.go @@ -166,11 +166,17 @@ func NewReader(src BlockSource, name string) (*Reader, error) { return nil, err } + // The hash ID is 4 raw bytes out of the file. Validate it before use: + // this runs before the CRC is compared below, so an unknown value here + // is reachable with no valid checksum at all. r.hashSize = r.header.HashID.Size() + if r.hashSize == 0 { + return nil, fmt.Errorf("%w: unknown hash id %q", fmtError, string(r.header.HashID[:])) + } r.header.BlockSize &= (1 << 24) - 1 if footBuf.Len() > 0 { - log.Panicf("footer size %d", footBuf.Len()) + return nil, fmt.Errorf("%w: trailing footer bytes: %d", fmtError, footBuf.Len()) } r.objectIDLen = int(r.footer.ObjOffset & ((1 << 5) - 1)) @@ -231,7 +237,7 @@ func (i *tableIter) nextInBlock(rec record) (bool, error) { } } if err != nil { - err = fmt.Errorf("block %c, off %d: %v", i.typ, i.blockOff, err) + err = fmt.Errorf("block %c, off %d: %w", i.typ, i.blockOff, err) } return ok, err } @@ -264,10 +270,13 @@ func (i *tableIter) Next(rec record) (bool, error) { // extractBlockSize returns the block size from the block header func extractBlockSize(block []byte, off uint64, version int) (typ byte, size uint32, err error) { if off == 0 { + if len(block) < headerSize(version) { + return 0, 0, fmtError + } block = block[headerSize(version):] } - if !isBlockType(block[0]) { + if len(block) < 4 || !isBlockType(block[0]) { return 0, 0, fmtError } @@ -334,7 +343,7 @@ func (i *tableIter) nextBlock() (bool, error) { nextBlockOff := i.blockOff + uint64(i.bi.br.fullBlockSize) br, err := i.r.newBlockReader(nextBlockOff, i.typ) if err != nil { - return false, fmt.Errorf("reftable: reading %c block at 0x%x: %v", i.typ, nextBlockOff, err) + return false, fmt.Errorf("reftable: reading %c block at 0x%x: %w", i.typ, nextBlockOff, err) } if br == nil { i.finished = true @@ -429,6 +438,11 @@ func (r *Reader) seek(rec record) (*tableIter, error) { if err != nil { return nil, err } + if tabIter == nil { + // No block of this type at the recorded offset; seekRecord turns a + // nil iterator into an empty one. + return nil, nil + } ok, err := r.seekLinear(tabIter, rec) if ok { @@ -448,28 +462,50 @@ func (r *Reader) seekIndexed(want record) (*tableIter, error) { LastKey: want.key(), } + // seekIndexed is only reached when IndexOffset is non-zero, so a nil + // iterator here means that offset does not address an index block. + if idxIter == nil { + return nil, fmt.Errorf("%w: index offset %d does not address an index block", + fmtError, r.offsets[want.typ()].IndexOffset) + } + ok, err := r.seekLinear(idxIter, wantIdx) if err != nil || !ok { return nil, err } - for { + // Every offset below is read out of the file, so each descent can be made + // to point anywhere, including back at the block we came from. Bound the + // walk: a real index tree has a handful of levels, since each one holds + // strictly fewer blocks than the level beneath it. + for depth := 0; ; depth++ { + if depth > maxIndexDepth { + return nil, fmt.Errorf("%w: index deeper than %d levels, probably cyclic", + fmtError, maxIndexDepth) + } + var rec indexRecord ok, err := idxIter.Next(&rec) - if !ok { - return nil, nil - } if err != nil { return nil, err } + if !ok { + return nil, nil + } tabIter, err := r.tabIterAt(rec.Offset, blockTypeAny) if err != nil { return nil, err } + if tabIter == nil { + // tabIterAt returns (nil, nil) when the offset is past EOF or + // otherwise unusable. The offset came from the file, so this is + // malformed input, not an empty result. + return nil, fmt.Errorf("%w: index entry points at offset %d, which is not a block", + fmtError, rec.Offset) + } - err = tabIter.bi.seek(want.key()) - if err != nil { + if err := tabIter.bi.seek(want.key()); err != nil { return nil, err } @@ -478,7 +514,8 @@ func (r *Reader) seekIndexed(want record) (*tableIter, error) { } if tabIter.typ != blockTypeIndex { - log.Panicf("got type %c following indexes", tabIter.typ) + return nil, fmt.Errorf("%w: index entry at offset %d has block type %c, want %c or %c", + fmtError, rec.Offset, tabIter.typ, want.typ(), blockTypeIndex) } idxIter = tabIter @@ -509,7 +546,8 @@ func (r *Reader) seekLinear(tabIter *tableIter, want record) (bool, error) { return false, err } if !ok { - panic("read from fresh block failed") + return false, fmt.Errorf("%w: block at offset %d yielded no records", + fmtError, tabIter.blockOff) } if rec.key() > wantKey { break @@ -590,6 +628,7 @@ func (i *indexedTableRefIter) Next(rec record) (bool, error) { } if bytes.Compare(ref.Value, i.oid) == 0 || bytes.Compare(ref.TargetValue, i.oid) == 0 { + ref.UpdateIndex += i.r.header.MinUpdateIndex return true, nil } } @@ -605,6 +644,9 @@ func (r *Reader) RefsFor(oid []byte) (*Iterator, error) { if err != nil { return nil, err } + if it == nil { + return &Iterator{&emptyIterator{}}, nil + } return &Iterator{&filteringRefIterator{ tab: r, oid: oid, @@ -614,12 +656,21 @@ func (r *Reader) RefsFor(oid []byte) (*Iterator, error) { } func (r *Reader) refsForIndexed(oid []byte) (*Iterator, error) { + // objectIDLen comes from the footer and can exceed the hash the caller + // passed, e.g. a 20 byte SHA-1 against a table declaring 31. + if r.objectIDLen > len(oid) { + return nil, fmt.Errorf("%w: table declares object id length %d, got a %d byte id", + fmtError, r.objectIDLen, len(oid)) + } want := &objRecord{HashPrefix: oid[:r.objectIDLen]} it, err := r.seek(want) if err != nil { return nil, err } + if it == nil { + return &Iterator{&emptyIterator{}}, nil + } got := objRecord{} ok, err := it.Next(&got) diff --git a/reader_index_validation_test.go b/reader_index_validation_test.go new file mode 100644 index 0000000..1f7680d --- /dev/null +++ b/reader_index_validation_test.go @@ -0,0 +1,94 @@ +package reftable + +import ( + "bytes" + "encoding/binary" + "errors" + "fmt" + "hash/crc32" + "math" + "testing" +) + +// Append a one-entry log-index root to an otherwise valid table. Keeping the +// original blocks and recalculating the footer CRC isolates index traversal +// from header validation. Unlike fixed byte flips, the fixture is independent +// of zlib output and changes in the writer's index layout. +func readerWithLogIndexTarget(t *testing.T, original *Reader, key string, target, rootOffset uint64) *Reader { + t.Helper() + bw := newBlockWriter(blockTypeIndex, make([]byte, 256), 0, original.hashSize) + if !bw.add(&indexRecord{LastKey: key, Offset: target}) { + t.Fatal("index record does not fit in fixture block") + } + raw := original.src.(*ByteBlockSource).Source + data := bytes.Clone(raw[:original.size]) + data = append(data, bw.finish()...) + footer := bytes.Clone(raw[original.size:]) + // LogIndexOffset is the fifth uint64 following the duplicated header. + binary.BigEndian.PutUint64(footer[headerSize(original.version)+4*8:], rootOffset) + binary.BigEndian.PutUint32(footer[len(footer)-4:], crc32.ChecksumIEEE(footer[:len(footer)-4])) + data = append(data, footer...) + r, err := NewReader(&ByteBlockSource{Source: data}, "index-target.ref") + if err != nil { + t.Fatalf("fixture must pass header/footer validation: %v", err) + } + t.Cleanup(r.Close) + return r +} + +func TestSeekLogRejectsInvalidIndexTargets(t *testing.T) { + var refs []RefRecord + var logs []LogRecord + for i := range 500 { + name := fmt.Sprintf("refs/heads/b%06d", i) + refs = append(refs, RefRecord{RefName: name, UpdateIndex: 1, Value: testHash(i)}) + logs = append(logs, LogRecord{ + RefName: name, UpdateIndex: 1, + New: testHash(i), Old: testHash(i), + Name: "n", Email: "e@x", Message: "m", + }) + } + _, original := constructTestTable(t, refs, logs, Config{BlockSize: 256}) + t.Cleanup(original.Close) + want := logs[0] + logOffset := original.offsets[blockTypeLog].Offset + if logOffset == 0 || original.offsets[blockTypeLog].IndexOffset == 0 { + t.Fatal("fixture must contain refs, logs, and a log index") + } + + for _, tc := range []struct { + name string + target uint64 + rootOffset uint64 + wantError bool + }{ + {"valid", logOffset, original.size, false}, + {"target_past_eof", math.MaxUint64, original.size, true}, + {"target_ref_block", 0, original.size, true}, + {"root_past_eof", logOffset, math.MaxUint64, true}, + {"root_log_block", logOffset, logOffset, true}, + // The fixture block sits at original.size, so pointing its single + // entry at that same offset makes the index reference itself. Without + // the maxIndexDepth bound in seekIndexed this descends forever. + {"cyclic_index", original.size, original.size, true}, + } { + t.Run(tc.name, func(t *testing.T) { + r := readerWithLogIndexTarget(t, original, want.key(), tc.target, tc.rootOffset) + it, err := r.SeekLog(want.RefName, want.UpdateIndex) + if tc.wantError { + if !errors.Is(err, fmtError) { + t.Fatalf("SeekLog error = %v, want format error", err) + } + return + } + if err != nil { + t.Fatal(err) + } + var got LogRecord + ok, err := it.NextLog(&got) + if err != nil || !ok || got.RefName != want.RefName || got.UpdateIndex != want.UpdateIndex || !bytes.Equal(got.New, want.New) { + t.Fatalf("valid index target: got (%+v, %v, %v)", got, ok, err) + } + }) + } +} diff --git a/record.go b/record.go index 95e46ec..d746f74 100644 --- a/record.go +++ b/record.go @@ -184,7 +184,7 @@ func (r *RefRecord) decode(buf []byte, key string, valType uint8, hashSize int) return } buf = buf[s:] - if len(buf) < int(tsize) { + if uint64(len(buf)) < tsize { return } @@ -329,6 +329,13 @@ func (r *objRecord) decode(buf []byte, prefix string, cnt3 uint8, hashSize int) return len(start) - len(buf), true } + // count is file-supplied. Each offset costs at least one byte of varint, + // so it cannot exceed what is left in the block; without this a 10 byte + // varint reserves gigabytes. + if count > uint64(len(buf)) { + return + } + r.Offsets = make([]uint64, 1, count) r.Offsets[0], n = getVarInt(buf) if n <= 0 { @@ -452,11 +459,14 @@ func decodeKey(buf []byte, prevKey string) (n int, key string, value uint8, ok b value = uint8(suffixLen & 0x7) suffixLen = suffixLen >> 3 - if int(suffixLen) > len(buf) { + // Compare as uint64: these come from varints, and int(v) is negative for + // v >= 2^63, which would let the guards pass and make() take a negative + // or absurd length. + if suffixLen > uint64(len(buf)) { return } - if int(prefixLen) > len(prevKey) { + if prefixLen > uint64(len(prevKey)) { return } @@ -582,7 +592,7 @@ func decodeString(buf []byte) (n int, val string, ok bool) { return } buf = buf[s:] - if len(buf) < int(nameLen) { + if uint64(len(buf)) < nameLen { return } val = string(buf[:nameLen]) diff --git a/reftable_test.go b/reftable_test.go index 190acbe..a730052 100644 --- a/reftable_test.go +++ b/reftable_test.go @@ -405,7 +405,9 @@ func TestTableIterRefLevel0(t *testing.T) { func TestTableSeekLogLevel1(t *testing.T) { // 25 * (50b + 60b) -> 13 blocks // 13 blocks -> 3 index blocks; not enough for another index level - testTableSeek(t, blockTypeLog, 25, 50, 256, 1, false) + // 25 records needs two index levels now that every level's final block + // is flushed; 7..20 records is the single-level band for this shape. + testTableSeek(t, blockTypeLog, 20, 50, 256, 1, false) } func TestTableLogBlocksUnaligned(t *testing.T) { diff --git a/review_regression_test.go b/review_regression_test.go new file mode 100644 index 0000000..6a026a4 --- /dev/null +++ b/review_regression_test.go @@ -0,0 +1,390 @@ +package reftable + +import ( + "bytes" + "errors" + "fmt" + "os" + "testing" +) + +func regressionStack(t *testing.T, storage Storage) *Stack { + t.Helper() + st, err := NewStack(storage, Config{}) + if err != nil { + t.Fatal(err) + } + st.disableAutoCompact = true + t.Cleanup(st.Close) + return st +} + +func regressionAddRef(t *testing.T, st *Stack, name string) { + t.Helper() + index := st.NextUpdateIndex() + err := st.Add(func(w *Writer) error { + w.SetLimits(index, index) + return w.AddRef(&RefRecord{ + RefName: name, UpdateIndex: index, + Value: bytes.Repeat([]byte{1}, 20), + }) + }) + if err != nil { + t.Fatal(err) + } +} + +func regressionRequireRef(t *testing.T, tab Table, name string) { + t.Helper() + it, err := tab.SeekRef(name) + if err != nil { + t.Fatal(err) + } + var ref RefRecord + ok, err := it.NextRef(&ref) + if err != nil || !ok || ref.RefName != name { + t.Fatalf("seek %q: got (%+v, %v, %v)", name, ref, ok, err) + } +} + +// Interleave two independent stacks deterministically, without goroutines or +// sleeps. Compaction calls Update after releasing the global manifest lock. +type regressionUpdateHook struct { + Storage + hook func() +} + +func (s *regressionUpdateHook) Update(name string) (AtomicWriter, error) { + if s.hook != nil { + hook := s.hook + s.hook = nil + hook() + } + return s.Storage.Update(name) +} + +func TestRegressionCompactionPreservesConcurrentAddition(t *testing.T) { + dir := t.TempDir() + storage := ®ressionUpdateHook{Storage: NewLocalStorage(dir)} + compactor := regressionStack(t, storage) + regressionAddRef(t, compactor, "refs/heads/a") + regressionAddRef(t, compactor, "refs/heads/b") + writer := regressionStack(t, NewLocalStorage(dir)) + interleaved := false + storage.hook = func() { + regressionAddRef(t, writer, "refs/heads/c") + regressionRequireRef(t, writer.Merged(), "refs/heads/c") + interleaved = true + } + if err := compactor.CompactAll(nil); err != nil { + t.Fatal(err) + } + if !interleaved { + t.Fatal("compaction did not execute the interleaved addition") + } + fresh := regressionStack(t, NewLocalStorage(dir)) + for _, name := range []string{"refs/heads/a", "refs/heads/b", "refs/heads/c"} { + regressionRequireRef(t, fresh.Merged(), name) + } +} + +func TestRegressionCompactionPreservesChangedPrefix(t *testing.T) { + dir := t.TempDir() + storage := ®ressionUpdateHook{Storage: NewLocalStorage(dir)} + compactor := regressionStack(t, storage) + for _, name := range []string{"a", "b", "c", "d"} { + regressionAddRef(t, compactor, "refs/heads/"+name) + } + other := regressionStack(t, NewLocalStorage(dir)) + storage.hook = func() { + if ok, err := other.compactRange(0, 1, nil); err != nil || !ok { + t.Fatalf("prefix compaction: %v, %v", ok, err) + } + } + if ok, err := compactor.compactRange(2, 3, nil); err != nil || !ok { + t.Fatalf("suffix compaction: %v, %v", ok, err) + } + fresh := regressionStack(t, NewLocalStorage(dir)) + if len(fresh.stack) != 2 { + t.Fatalf("got %d tables, want both compacted ranges", len(fresh.stack)) + } + for _, name := range []string{"a", "b", "c", "d"} { + regressionRequireRef(t, fresh.Merged(), "refs/heads/"+name) + } +} + +func TestRegressionCompactionLockContention(t *testing.T) { + dir := t.TempDir() + storage := ®ressionUpdateHook{Storage: NewLocalStorage(dir)} + st := regressionStack(t, storage) + regressionAddRef(t, st, "refs/heads/a") + regressionAddRef(t, st, "refs/heads/b") + storage.hook = func() { + lock, err := storage.LockForWrite(listFileName) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { lock.Close() }) + } + if ok, err := st.compactRange(0, 1, nil); ok || err != nil { + t.Fatalf("contention: got (%v, %v), want (false, nil)", ok, err) + } + lock, err := storage.LockForWrite(listFileName) + if err == nil { + lock.Close() + t.Fatal("compaction removed the competing writer's lock") + } + if !os.IsExist(err) { + t.Fatal(err) + } + regressionRequireRef(t, st.Merged(), "refs/heads/a") + regressionRequireRef(t, st.Merged(), "refs/heads/b") +} + +func TestRegressionReloadFailurePreservesExistingReaders(t *testing.T) { + st := regressionStack(t, NewLocalStorage(t.TempDir())) + regressionAddRef(t, st, "refs/heads/a") + regressionRequireRef(t, st.Merged(), "refs/heads/a") + // The first reader is borrowed from the live stack; the second open fails. + if err := st.reloadOnce([]string{st.stack[0].Name(), "missing.ref"}, true); err == nil { + t.Fatal("expected missing-table error") + } + regressionRequireRef(t, st.Merged(), "refs/heads/a") +} + +func TestRegressionCleanEmptyStack(t *testing.T) { + st := regressionStack(t, NewLocalStorage(t.TempDir())) + defer func() { + if p := recover(); p != nil { + t.Errorf("empty-stack operation panicked: %v", p) + } + }() + if err := st.Clean(); err != nil { + t.Fatal(err) + } + if st.Merged().MinUpdateIndex() != 0 || st.Merged().MaxUpdateIndex() != 0 { + t.Fatal("empty stack must have zero update-index bounds") + } + if err := st.CompactAll(&LogExpirationConfig{Time: 1}); err != nil { + t.Fatal(err) + } +} + +func TestRegressionIndexedRefsForUpdateIndex(t *testing.T) { + var buf bytes.Buffer + w, err := NewWriter(&buf, &Config{BlockSize: 256}) + if err != nil { + t.Fatal(err) + } + w.SetLimits(100, 200) + oid := bytes.Repeat([]byte{1}, 20) + for i := 0; i < 100; i++ { + if err := w.AddRef(&RefRecord{ + RefName: fmt.Sprintf("refs/heads/%04d", i), + UpdateIndex: 150, Value: oid, + }); err != nil { + t.Fatal(err) + } + } + if err := w.Close(); err != nil { + t.Fatal(err) + } + if w.Stats.ObjStats.Blocks == 0 { + t.Fatal("fixture must contain an object index") + } + r, err := NewReader(&ByteBlockSource{Source: buf.Bytes()}, "indexed.ref") + if err != nil { + t.Fatal(err) + } + defer r.Close() + it, err := r.RefsFor(oid) + if err != nil { + t.Fatal(err) + } + count := 0 + for { + var ref RefRecord + ok, err := it.NextRef(&ref) + if err != nil { + t.Fatal(err) + } + if !ok { + break + } + if ref.UpdateIndex != 150 { + t.Fatalf("%s: UpdateIndex = %d, want 150", ref.RefName, ref.UpdateIndex) + } + count++ + } + if count != 100 { + t.Fatalf("got %d references, want 100", count) + } +} + +func TestRegressionCorruptBlockReturnsError(t *testing.T) { + var buf bytes.Buffer + w, err := NewWriter(&buf, &Config{}) + if err != nil { + t.Fatal(err) + } + w.SetLimits(1, 1) + if err := w.AddRef(&RefRecord{RefName: "refs/heads/a", UpdateIndex: 1, Value: bytes.Repeat([]byte{1}, 20)}); err != nil { + t.Fatal(err) + } + if err := w.Close(); err != nil { + t.Fatal(err) + } + for _, size := range []uint32{0, 1, (1 << 24) - 1} { + t.Run(fmt.Sprintf("size_%d", size), func(t *testing.T) { + defer func() { + if p := recover(); p != nil { + t.Errorf("malformed block caused panic instead of error: %v", p) + } + }() + data := bytes.Clone(buf.Bytes()) + // The block header is not covered by the footer CRC. + putU24(data[headerSize(1)+1:], size) + r, err := NewReader(&ByteBlockSource{Source: data}, "corrupt.ref") + if err != nil { + return // Rejecting corruption at open time is also correct. + } + defer r.Close() + it, err := r.SeekRef("") + if err == nil { + var ref RefRecord + _, err = it.NextRef(&ref) + } + if err == nil { + t.Fatal("malformed block was accepted") + } + }) + } +} + +// Model Commit's publication/durability boundary: the rename succeeds, but +// directory fsync reports an error. Committed remains true through embedding. +type regressionCommitErrorWriter struct { + AtomicWriter + err error +} + +func (w *regressionCommitErrorWriter) Commit() error { + if err := w.AtomicWriter.Commit(); err != nil { + return err + } + return w.err +} + +type regressionCommitErrorStorage struct { + Storage + err error +} + +func (s *regressionCommitErrorStorage) LockForWrite(name string) (AtomicWriter, error) { + w, err := s.Storage.LockForWrite(name) + if err != nil { + return nil, err + } + if name == listFileName { + return ®ressionCommitErrorWriter{AtomicWriter: w, err: s.err}, nil + } + return w, nil +} + +// Fail the table's directory sync, rather than the manifest's. The table +// has been renamed but is not yet reachable through tables.list. +type regressionTableSyncErrorStorage struct { + Storage + err error +} + +func (s *regressionTableSyncErrorStorage) Update(name string) (AtomicWriter, error) { + w, err := s.Storage.Update(name) + if err != nil || s.err == nil { + return w, err + } + return ®ressionCommitErrorWriter{AtomicWriter: w, err: s.err}, nil +} + +func TestRegressionTableSyncFailureRemovesUnreferencedTable(t *testing.T) { + for _, existing := range []bool{false, true} { + t.Run(fmt.Sprintf("existing=%v", existing), func(t *testing.T) { + dir := t.TempDir() + storage := ®ressionTableSyncErrorStorage{Storage: NewLocalStorage(dir)} + st := regressionStack(t, storage) + if existing { + regressionAddRef(t, st, "refs/heads/existing") + } + before, err := storage.ReadDir() + if err != nil { + t.Fatal(err) + } + syncErr := errors.New("injected table sync failure after rename") + storage.err = syncErr + index := st.NextUpdateIndex() + err = st.Add(func(w *Writer) error { + w.SetLimits(index, index) + return w.AddRef(&RefRecord{RefName: "refs/heads/new", UpdateIndex: index, Value: testHash(1)}) + }) + if !errors.Is(err, syncErr) { + t.Fatalf("Add error = %v, want injected sync error", err) + } + after, err := storage.ReadDir() + if err != nil { + t.Fatal(err) + } + if len(after) != len(before) { + t.Fatalf("failed addition left %d directory entries, want %d", len(after), len(before)) + } + for i := range before { + if after[i].Name() != before[i].Name() { + t.Fatalf("directory entry changed: %s -> %s", before[i].Name(), after[i].Name()) + } + } + fresh := regressionStack(t, NewLocalStorage(dir)) + if existing { + regressionRequireRef(t, fresh.Merged(), "refs/heads/existing") + } + if ref, err := ReadRef(fresh.Merged(), "refs/heads/new"); err != nil || ref != nil { + t.Fatalf("failed addition became visible: %+v, %v", ref, err) + } + storage.err = nil + regressionAddRef(t, st, "refs/heads/new") + regressionRequireRef(t, st.Merged(), "refs/heads/new") + }) + } +} + +func TestRegressionCompactedManifestSurvivesSyncError(t *testing.T) { + dir := t.TempDir() + storage := ®ressionCommitErrorStorage{Storage: NewLocalStorage(dir)} + st := regressionStack(t, storage) + regressionAddRef(t, st, "refs/heads/a") + regressionAddRef(t, st, "refs/heads/b") + syncErr := errors.New("injected compaction directory sync failure") + storage.err = syncErr + requirePostCommitError(t, st.CompactAll(nil), syncErr) + fresh := regressionStack(t, NewLocalStorage(dir)) + if len(fresh.stack) != 1 { + t.Fatalf("got %d tables, want published compacted table", len(fresh.stack)) + } + regressionRequireRef(t, fresh.Merged(), "refs/heads/a") + regressionRequireRef(t, fresh.Merged(), "refs/heads/b") +} + +func TestRegressionPublishedManifestSurvivesSyncError(t *testing.T) { + dir := t.TempDir() + syncErr := errors.New("injected directory sync failure after rename") + st := regressionStack(t, ®ressionCommitErrorStorage{ + Storage: NewLocalStorage(dir), err: syncErr, + }) + err := st.Add(func(w *Writer) error { + w.SetLimits(1, 1) + return w.AddRef(&RefRecord{RefName: "refs/heads/a", UpdateIndex: 1, Value: bytes.Repeat([]byte{1}, 20)}) + }) + requirePostCommitError(t, err, syncErr) + // The manifest is already visible. Its tables must not be rolled back, + // even though the caller was told that durability could not be confirmed. + fresh := regressionStack(t, NewLocalStorage(dir)) + regressionRequireRef(t, fresh.Merged(), "refs/heads/a") +} diff --git a/stack.go b/stack.go index 5b1bb37..50341a9 100644 --- a/stack.go +++ b/stack.go @@ -12,12 +12,12 @@ import ( "bytes" "errors" "fmt" - "log" "math" - "math/rand" + "math/rand/v2" "os" "path/filepath" "reflect" + "slices" "strings" "time" ) @@ -79,6 +79,21 @@ func (st *Stack) String() string { return fmt.Sprintf("%v", nms) } +// validateTableName rejects manifest entries that would escape the reftable +// directory. Storage implementations join these names onto a base directory, +// and filepath.Join cleans "..", so an unvalidated name from tables.list can +// address — and Remove — arbitrary files. Table names are always plain +// filenames, so refusing separators and dot components is sufficient. +func validateTableName(name string) error { + if name == "" || name == "." || name == ".." { + return fmt.Errorf("%w: invalid table name %q", fmtError, name) + } + if strings.ContainsAny(name, `/\`) || strings.Contains(name, "\x00") { + return fmt.Errorf("%w: table name %q must be a plain filename", fmtError, name) + } + return nil +} + func (st *Stack) readNames() ([]string, error) { bs, err := st.storage.OpenBlockSource(listFileName) if errors.Is(err, os.ErrNotExist) { @@ -90,13 +105,6 @@ func (st *Stack) readNames() ([]string, error) { defer bs.Close() data, err := bs.ReadBlock(0, int(bs.Size())) - if err != nil { - log.Printf("err %v %s %d", err, data, bs.Size()) - return nil, err - } - if errors.Is(err, os.ErrNotExist) { - return nil, nil - } if err != nil { return nil, err } @@ -104,9 +112,14 @@ func (st *Stack) readNames() ([]string, error) { var res []string for _, l := range lines { - if len(l) > 0 { - res = append(res, string(l)) + if len(l) == 0 { + continue + } + name := string(l) + if err := validateTableName(name); err != nil { + return nil, err } + res = append(res, name) } return res, nil @@ -146,14 +159,16 @@ func (st *Stack) reloadOnce(names []string, reuseOpen bool) error { cur[r.Name()] = r } - var newTables []*Reader + var newTables, opened []*Reader + retained := make(map[string]bool, len(names)) defer func() { - for _, t := range newTables { + for _, t := range opened { t.Close() } }() for _, name := range names { + retained[name] = true rd := cur[name] if reuseOpen && rd != nil { delete(cur, name) @@ -165,21 +180,36 @@ func (st *Stack) reloadOnce(names []string, reuseOpen bool) error { rd, err = NewReader(bs, name) if err != nil { - return fmt.Errorf("NewReader(%s): %v", name, err) + bs.Close() + return fmt.Errorf("NewReader(%s): %w", name, err) } + opened = append(opened, rd) } newTables = append(newTables, rd) } - // success. Swap. + var tabs []Table + for _, r := range newTables { + tabs = append(tabs, r) + } + merged, err := NewMerged(tabs, st.cfg.HashID) + if err != nil { + return err + } + merged.suppressDeletions = true + + // Only transfer ownership once the entire replacement is valid. st.stack = newTables - newTables = nil + st.merged = merged + opened = nil for _, old := range cur { old.Close() // On windows, we may only be able to close after // closing file handles. - st.storage.Remove(old.Name()) + if !retained[old.Name()] { + st.storage.Remove(old.Name()) + } } return nil } @@ -199,7 +229,7 @@ func (st *Stack) reload(reuseOpen bool) error { } err = st.reloadOnce(names, reuseOpen) if err == nil { - break + return nil } if !errors.Is(err, os.ErrNotExist) { return err @@ -214,28 +244,63 @@ func (st *Stack) reload(reuseOpen bool) error { } // compaction changed names; back off and retry. - delay = 2*delay + time.Millisecond*time.Duration(1+rand.Intn(2)) + delay = 2*delay + time.Millisecond*time.Duration(1+rand.IntN(2)) time.Sleep(delay) } - var tabs []Table - for _, r := range st.stack { - tabs = append(tabs, r) - } + return ErrReloadTimeout +} - m, err := NewMerged(tabs, st.cfg.HashID) - if err != nil { +// ErrLockFailure means a write could not proceed before publishing its +// manifest, for example because a lock was contended or the stack was stale. +// Callers may retry the transaction when errors.Is(err, ErrLockFailure). +// Errors after publication never match this sentinel. +var ErrLockFailure = errors.New("reftable: lock failure") + +// ErrReloadTimeout means no consistent stack snapshot could be loaded before +// the reload deadline. It does not mean that a preceding write was aborted. +var ErrReloadTimeout = errors.New("reftable: stack reload timed out") + +// ErrPostCommit means the manifest was published, but durability confirmation, +// reloading, or maintenance failed afterward. Do not replay the transaction. +// Use errors.As to inspect the Cause of the accompanying *PostCommitError. +var ErrPostCommit = errors.New("reftable: error after manifest publication") + +// PostCommitError reports a failure after a manifest was published. The update +// is visible, though durability may be uncertain and the Stack may still hold +// its previous snapshot. Reopen the stack to inspect the current state; do not +// blindly retry the write. +// +// Cause is deliberately not part of the Unwrap chain: a lock error during +// post-commit maintenance must not classify a committed write as retryable. +// Callers can inspect it explicitly with errors.Is or errors.As after checking +// the publication status. +type PostCommitError struct { + Cause error +} + +func (e *PostCommitError) Error() string { + return fmt.Sprintf("%v: %v", ErrPostCommit, e.Cause) +} + +func (e *PostCommitError) Unwrap() error { return ErrPostCommit } + +func postCommitError(err error) error { + if err == nil { + return nil + } + // errors.As, not a type assertion: callers reach here through + // errors.Join and fmt.Errorf("%w", ...), so the PostCommitError is + // frequently wrapped rather than the top-level value. A bare assertion + // misses those and wraps a second time, and the whole point of this type + // is that a committed write is not reclassified as retryable. + var pce *PostCommitError + if errors.As(err, &pce) { return err } - m.suppressDeletions = true - st.merged = m - return nil + return &PostCommitError{Cause: err} } -// ErrLockFailure is returned for failed writes. On a failed write, -// the stack is reloaded, so the transaction may be retried. -var ErrLockFailure = errors.New("reftable: lock failure") - func (st *Stack) UpToDate() (bool, error) { names, err := st.readNames() if err != nil { @@ -254,32 +319,41 @@ func (st *Stack) UpToDate() (bool, error) { return true, nil } -// Add a new reftable to stack, transactionally. +// Add a new reftable to stack, transactionally. ErrPostCommit means the update +// was published but a later step failed; the callback must not be replayed. func (st *Stack) Add(write func(w *Writer) error) error { - if err := st.add(write); err != nil { - if err == ErrLockFailure { + published, err := st.add(write) + if err != nil { + if errors.Is(err, ErrLockFailure) { st.reload(true) } return err } if !st.disableAutoCompact { - return st.AutoCompact() + err = st.AutoCompact() + if published { + // The addition committed, even if maintenance cannot lock or + // reload the stack. Do not advertise a retryable write failure. + return postCommitError(err) + } + return err } return nil } -func (st *Stack) add(write func(w *Writer) error) error { +func (st *Stack) add(write func(w *Writer) error) (bool, error) { tr, err := st.NewAddition() if err != nil { - return err + return false, err } defer tr.Close() if err := tr.Add(write); err != nil { - return err + return false, err } - return tr.Commit() + err = tr.Commit() + return tr.lockFile.Committed(), err } // Addition is a transaction that adds new tables to the top of the @@ -360,6 +434,12 @@ func (tr *Addition) Add(write func(w *Writer) error) error { } if err := tab.Commit(); err != nil { + // The table rename may have succeeded before directory sync failed. + // It is not in the manifest yet, so remove it explicitly: Close must + // preserve published files, and this one is not tracked by tr yet. + if tab.Committed() { + return errors.Join(err, tr.stack.storage.Remove(dest)) + } return err } @@ -374,10 +454,12 @@ func (tr *Addition) Close() { for _, nm := range tr.newTables { tr.stack.storage.Remove(nm) } + tr.newTables = nil tr.lockFile.Close() } -// Commit commits the changes to the database, releasing the lock. +// Commit commits the changes to the database, releasing the lock. It returns +// ErrPostCommit if the manifest was published but a subsequent step failed. func (tr *Addition) Commit() error { if len(tr.newTables) == 0 { // Nothing to be done. @@ -389,13 +471,15 @@ func (tr *Addition) Commit() error { return err } - if err := tr.lockFile.Commit(); err != nil { + err := tr.lockFile.Commit() + if err != nil && !tr.lockFile.Committed() { tr.Close() return err } + // Rename may succeed even when the following directory sync fails. + // Published tables belong to the manifest and must survive cleanup. tr.newTables = nil - - return tr.stack.reload(true) + return postCommitError(errors.Join(err, tr.stack.reload(true))) } func (s *Stack) checkAddition(tabname string) error { @@ -408,7 +492,8 @@ func (s *Stack) checkAddition(tabname string) error { } r, err := NewReader(bs, tabname) if err != nil { - return err + bs.Close() + return fmt.Errorf("NewReader(%s): %w", tabname, err) } defer r.Close() it, err := r.SeekRef("") @@ -432,11 +517,14 @@ func (s *Stack) checkAddition(tabname string) error { return validateRefRecordAddition(s.Merged(), recs) } -// non-deterministic random generator. -var randomRandom = rand.New(rand.NewSource(time.Now().UnixNano())) - +// formatName builds the filename for a table covering [min, max]. The random +// suffix keeps names unique when several tables cover the same update-index +// range. rand/v2's top-level source is safe for concurrent use, which matters +// because one process commonly holds a Stack per repository and writes to them +// from different goroutines; the old package-level *rand.Rand was not, and +// torn reads produced duplicate names. func formatName(min, max uint64) string { - return fmt.Sprintf("0x%012x-0x%012x-%08x", min, max, randomRandom.Uint32()) + return fmt.Sprintf("0x%012x-0x%012x-%08x", min, max, rand.Uint32()) } // NextUpdateIndex returns the update index at which to write the next table. @@ -567,7 +655,7 @@ func (st *Stack) compactRangeStats(first, last int, expiration *LogExpirationCon } func (st *Stack) compactRange(first, last int, expiration *LogExpirationConfig) (bool, error) { - if first >= last && expiration == nil { + if first > last || (first == last && expiration == nil) { return true, nil } st.Stats.Attempts++ @@ -580,9 +668,7 @@ func (st *Stack) compactRange(first, last int, expiration *LogExpirationConfig) return false, err } - defer func() { - lock.Close() - }() + defer lock.Close() if ok, err := st.UpToDate(); !ok || err != nil { return false, err @@ -622,64 +708,77 @@ func (st *Stack) compactRange(first, last int, expiration *LogExpirationConfig) if err != nil { return false, err } + published := false if tmpTable != nil { - defer tmpTable.Close() + defer func() { + if !published && tmpTable.Committed() { + st.storage.Remove(tmpTable.Name()) + } + tmpTable.Close() + }() } lock, err = st.storage.LockForWrite(listFileName) + if errors.Is(err, os.ErrExist) { + return false, nil + } if err != nil { return false, err } - defer lock.Close() + // Other writers can append or compact unrelated tables while the global + // lock is released. Replace only our still-contiguous range in the latest + // manifest, preserving every change outside it. + current, err := st.readNames() + if err != nil { + return false, err + } + if len(deleteOnSuccess) == 0 { + // Guaranteed by the first > last check above; keep the slice + // access below honest if that guard is ever relaxed. + return false, nil + } + start := slices.Index(current, deleteOnSuccess[0]) + end := start + len(deleteOnSuccess) + if start < 0 || end > len(current) || !slices.Equal(current[start:end], deleteOnSuccess) { + return false, nil + } + var names []string + names = append(names, current[:start]...) if tmpTable != nil { if err := tmpTable.Commit(); err != nil { return false, err } - } - - var names []string - for i := range first { - names = append(names, st.stack[i].name) - } - - if tmpTable != nil { names = append(names, tmpTable.Name()) } - - for i := last + 1; i < len(st.stack); i++ { - names = append(names, st.stack[i].name) - } + names = append(names, current[end:]...) if _, err := lock.Write([]byte(strings.Join(names, "\n"))); err != nil { - if tmpTable != nil { - os.Remove(tmpTable.Name()) - } return false, err } - if err := lock.Commit(); err != nil { - if tmpTable != nil { - os.Remove(tmpTable.Name()) - } + err = lock.Commit() + published = err == nil || lock.Committed() + if !published { return false, err } - for _, nm := range deleteOnSuccess { - if tmpTable != nil && nm != tmpTable.Name() { - // reflog expiry might cause us to reopen a - // new file with the same name. - os.Remove(nm) + // Reload closes and removes superseded tables through Storage. A sync + // failure after publication must not roll back the replacement table. + reloadErr := st.reload(expiration == nil) + if reloadErr != nil { + // reloadOnce never reached its cleanup, so the tables we just + // replaced are unreferenced by the manifest but still on disk. + // Nothing else collects them; drop them here. + for _, nm := range deleteOnSuccess { + if tmpTable != nil && nm == tmpTable.Name() { + // Reflog expiry can reuse the name we just published. + continue + } + st.storage.Remove(nm) } } - - // If we expire log entries on a full compaction we write a - // table with the same the (min,max) update index, but we have - // to read from disk again. - if err := st.reload(expiration == nil); err != nil { - return true, fmt.Errorf("reload: %w", err) - } - return true, err + return true, postCommitError(errors.Join(err, reloadErr)) } func (st *Stack) tableSizesForCompaction() []uint64 { @@ -834,7 +933,8 @@ func (st *Stack) Clean() error { rd, err := NewReader(bs, name) if err != nil { - return fmt.Errorf("NewReader(%s): %v", name, err) + bs.Close() + return fmt.Errorf("NewReader(%s): %w", name, err) } cur := rd.MaxUpdateIndex() diff --git a/stack_commit_test.go b/stack_commit_test.go new file mode 100644 index 0000000..1b43440 --- /dev/null +++ b/stack_commit_test.go @@ -0,0 +1,225 @@ +package reftable + +import ( + "errors" + "fmt" + "testing" +) + +func requirePostCommitError(t *testing.T, err, cause error) { + t.Helper() + if !errors.Is(err, ErrPostCommit) { + t.Fatalf("got %v, want ErrPostCommit", err) + } + if errors.Is(err, ErrLockFailure) { + t.Fatalf("published update was classified as retryable: %v", err) + } + var published *PostCommitError + if !errors.As(err, &published) || !errors.Is(published.Cause, cause) { + t.Fatalf("got %v, want explicit post-commit cause %v", err, cause) + } +} + +// Inject reload or maintenance failures only after the real manifest rename. +// No sleep is needed to model a reload exhausting its retry deadline. +type publicationFailureStorage struct { + Storage + afterPublication func() + readErr error + manifestLocks int + failLockAt int +} + +func (s *publicationFailureStorage) OpenBlockSource(name string) (BlockSource, error) { + if name == listFileName && s.readErr != nil { + return nil, s.readErr + } + return s.Storage.OpenBlockSource(name) +} + +func (s *publicationFailureStorage) LockForWrite(name string) (AtomicWriter, error) { + if name == listFileName { + s.manifestLocks++ + if s.manifestLocks == s.failLockAt { + return nil, fmt.Errorf("injected maintenance failure: %w", ErrLockFailure) + } + } + w, err := s.Storage.LockForWrite(name) + if err != nil || name != listFileName || s.afterPublication == nil { + return w, err + } + return &publicationHookWriter{AtomicWriter: w, hook: s.afterPublication}, nil +} + +type publicationHookWriter struct { + AtomicWriter + hook func() +} + +func (w *publicationHookWriter) Commit() error { + if err := w.AtomicWriter.Commit(); err != nil { + return err + } + w.hook() + return nil +} + +func TestPublishedReloadFailureIsNotRetryable(t *testing.T) { + for _, operation := range []string{"Add", "Addition.Commit", "CompactAll"} { + for _, cause := range []error{ErrReloadTimeout, fmt.Errorf("storage: %w", ErrLockFailure)} { + t.Run(fmt.Sprintf("%s/%v", operation, cause), func(t *testing.T) { + dir := t.TempDir() + storage := &publicationFailureStorage{Storage: NewLocalStorage(dir)} + st := regressionStack(t, storage) + regressionAddRef(t, st, "refs/heads/a") + if operation == "CompactAll" { + regressionAddRef(t, st, "refs/heads/b") + } + storage.afterPublication = func() { storage.readErr = cause } + calls := 0 + write := func(w *Writer) error { + calls++ + index := st.NextUpdateIndex() + w.SetLimits(index, index) + return w.AddRef(&RefRecord{RefName: "refs/heads/b", UpdateIndex: index, Value: testHash(2)}) + } + var err error + switch operation { + case "Add": + // Typical client retry policy must not replay the callback. + for attempts := 0; attempts < 2; attempts++ { + err = st.Add(write) + if !errors.Is(err, ErrLockFailure) { + break + } + } + case "Addition.Commit": + tr, openErr := st.NewAddition() + if openErr != nil { + t.Fatal(openErr) + } + defer tr.Close() + if err := tr.Add(write); err != nil { + t.Fatal(err) + } + err = tr.Commit() + case "CompactAll": + err = st.CompactAll(nil) + } + requirePostCommitError(t, err, cause) + if operation != "CompactAll" && calls != 1 { + t.Fatalf("write callback ran %d times, want once", calls) + } + storage.readErr = nil + storage.afterPublication = nil + fresh := regressionStack(t, NewLocalStorage(dir)) + regressionRequireRef(t, fresh.Merged(), "refs/heads/a") + regressionRequireRef(t, fresh.Merged(), "refs/heads/b") + if fresh.NextUpdateIndex() != 3 { + t.Fatalf("unexpected update-index advancement: %d", fresh.NextUpdateIndex()) + } + if operation == "CompactAll" && len(fresh.stack) != 1 { + t.Fatal("compacted manifest was not published") + } + }) + } + } +} + +func TestAutoCompactionFailureAfterAdditionIsNotRetryable(t *testing.T) { + storage := &publicationFailureStorage{Storage: NewLocalStorage(t.TempDir())} + st := regressionStack(t, storage) + regressionAddRef(t, st, "refs/heads/a") + st.disableAutoCompact = false + storage.manifestLocks = 0 + storage.failLockAt = 2 // First lock publishes the addition; second is maintenance. + err := st.Add(func(w *Writer) error { + w.SetLimits(2, 2) + return w.AddRef(&RefRecord{RefName: "refs/heads/b", UpdateIndex: 2, Value: testHash(2)}) + }) + requirePostCommitError(t, err, ErrLockFailure) + regressionRequireRef(t, st.Merged(), "refs/heads/b") +} + +func TestLockFailureBeforePublicationRemainsRetryable(t *testing.T) { + storage := NewLocalStorage(t.TempDir()) + st := regressionStack(t, storage) + lock, err := storage.LockForWrite(listFileName) + if err != nil { + t.Fatal(err) + } + defer lock.Close() + called := false + err = st.Add(func(w *Writer) error { called = true; return nil }) + if called || !errors.Is(err, ErrLockFailure) || errors.Is(err, ErrPostCommit) { + t.Fatalf("pre-publication contention: callback=%v, error=%v", called, err) + } +} + +func TestEmptyAdditionDoesNotReportPublication(t *testing.T) { + storage := &publicationFailureStorage{Storage: NewLocalStorage(t.TempDir())} + st := regressionStack(t, storage) + regressionAddRef(t, st, "refs/heads/a") + regressionAddRef(t, st, "refs/heads/b") + st.disableAutoCompact = false + storage.manifestLocks = 0 + storage.failLockAt = 2 + err := st.Add(func(w *Writer) error { return nil }) + if !errors.Is(err, ErrLockFailure) || errors.Is(err, ErrPostCommit) { + t.Fatalf("nothing was published: got %v, want retryable maintenance error", err) + } +} + +// postCommitError classifies whether a write survived publication, so it must +// recognise a PostCommitError that arrives wrapped. It reaches this function +// through errors.Join and fmt.Errorf("%w", ...) on the compaction and addition +// paths, and a bare type assertion misses both and wraps a second time. +func TestPostCommitErrorDoesNotDoubleWrap(t *testing.T) { + if got := postCommitError(nil); got != nil { + t.Errorf("postCommitError(nil) = %v, want nil", got) + } + + cause := errors.New("directory sync failed") + once := postCommitError(cause) + + var first *PostCommitError + if !errors.As(once, &first) { + t.Fatalf("postCommitError(%v) is not a *PostCommitError", cause) + } + if !errors.Is(once, ErrPostCommit) { + t.Fatal("a post-commit error must satisfy errors.Is(err, ErrPostCommit)") + } + + for _, tc := range []struct { + name string + err error + }{ + {"bare", once}, + {"joined_with_nil", errors.Join(once, nil)}, + {"joined_with_error", errors.Join(once, errors.New("reload failed"))}, + {"fmt_wrapped", fmt.Errorf("compact: %w", once)}, + } { + t.Run(tc.name, func(t *testing.T) { + got := postCommitError(tc.err) + + if !errors.Is(got, ErrPostCommit) { + t.Fatalf("lost the post-commit classification: %v", got) + } + + var outer *PostCommitError + if !errors.As(got, &outer) { + t.Fatalf("no *PostCommitError in %v", got) + } + var inner *PostCommitError + if errors.As(outer.Cause, &inner) { + t.Errorf("wrapped a second time: Cause is itself a *PostCommitError (%v)", outer.Cause) + } + // Note: Unwrap returns ErrPostCommit, not Cause, so the + // underlying error is reachable through the exported field + // rather than errors.Is. + if !errors.Is(outer.Cause, cause) { + t.Errorf("Cause = %v, want the original %v", outer.Cause, cause) + } + }) + } +} diff --git a/stack_concurrent_test.go b/stack_concurrent_test.go new file mode 100644 index 0000000..6d93801 --- /dev/null +++ b/stack_concurrent_test.go @@ -0,0 +1,125 @@ +package reftable + +import ( + "fmt" + "os" + "path/filepath" + "sync" + "testing" +) + +// A process that serves several repositories holds one Stack per repository +// and writes to them from different goroutines. Each Stack is used by a single +// goroutine — Stack itself is not safe for concurrent use — but everything a +// Stack touches that is shared across instances must be. +// +// Run under -race: this is what catches shared writer state such as the +// package-level RNG behind formatName. +func TestConcurrentStacksInSeparateRepos(t *testing.T) { + const repos = 8 + const writesPerRepo = 6 + + dirs := make([]string, repos) + root := t.TempDir() + for i := range dirs { + dirs[i] = filepath.Join(root, fmt.Sprintf("repo%d", i)) + if err := os.MkdirAll(dirs[i], 0o755); err != nil { + t.Fatal(err) + } + } + + var wg sync.WaitGroup + errs := make([]error, repos) + start := make(chan struct{}) + + for i := range repos { + wg.Add(1) + go func() { + defer wg.Done() + <-start // maximise overlap + + st, err := NewStack(NewLocalStorage(dirs[i]), Config{}) + if err != nil { + errs[i] = fmt.Errorf("NewStack: %w", err) + return + } + defer st.Close() + + for j := range writesPerRepo { + name := fmt.Sprintf("refs/heads/repo%d-%d", i, j) + err := st.Add(func(w *Writer) error { + idx := st.NextUpdateIndex() + w.SetLimits(idx, idx) + return w.AddRef(&RefRecord{ + RefName: name, + UpdateIndex: idx, + Value: testHash(i*writesPerRepo + j), + }) + }) + if err != nil { + errs[i] = fmt.Errorf("Add(%s): %w", name, err) + return + } + } + + // Every ref this goroutine wrote must be readable from its stack. + for j := range writesPerRepo { + name := fmt.Sprintf("refs/heads/repo%d-%d", i, j) + rec, err := ReadRef(st.Merged(), name) + if err != nil { + errs[i] = fmt.Errorf("ReadRef(%s): %w", name, err) + return + } + if rec == nil { + errs[i] = fmt.Errorf("ref %s missing after concurrent writes", name) + return + } + } + }() + } + + close(start) + wg.Wait() + + for i, err := range errs { + if err != nil { + t.Errorf("repo%d: %v", i, err) + } + } +} + +// Table names must be unique across concurrently-created additions: they are +// the filenames the manifest refers to, and a collision loses a table. +func TestConcurrentTableNamesAreUnique(t *testing.T) { + const goroutines = 16 + const each = 64 + + var mu sync.Mutex + seen := map[string]bool{} + dup := "" + + var wg sync.WaitGroup + for range goroutines { + wg.Add(1) + go func() { + defer wg.Done() + local := make([]string, 0, each) + for j := range each { + local = append(local, formatName(uint64(j), uint64(j))) + } + mu.Lock() + defer mu.Unlock() + for _, n := range local { + if seen[n] && dup == "" { + dup = n + } + seen[n] = true + } + }() + } + wg.Wait() + + if dup != "" { + t.Errorf("duplicate table name generated concurrently: %s", dup) + } +} diff --git a/storage.go b/storage.go index 530eab4..cd47e94 100644 --- a/storage.go +++ b/storage.go @@ -10,11 +10,23 @@ import ( ) // AtomicWriter is an abstraction for a {write to temp, close, rename} -// file sink. +// file sink. Close aborts unpublished output and must be idempotent: a +// second Close must not remove a path it no longer owns, because callers +// legitimately Close the same writer twice on error paths. type AtomicWriter interface { io.WriteCloser + + // Name returns the basename this writer's output currently occupies: + // the temporary name before Commit, the final name after a successful + // one. It is stable across Close, so `Remove(w.Name())` after an + // aborted Close never addresses the final path. Name() string Commit() error + + // Committed reports whether the new file has been published, i.e. the + // rename succeeded, even if Commit then returned a durability error. + // Close must not remove a published file. Note this is narrower than + // "the writer was closed": an aborted writer reports false. Committed() bool } @@ -36,28 +48,38 @@ type Storage interface { type fileWriter struct { finalName string + tempName string + committed bool + aborted bool *os.File } func (fw *fileWriter) Committed() bool { - return fw.File == nil + return fw.committed } func (fw *fileWriter) Name() string { - if fw.File != nil { - return filepath.Base(fw.File.Name()) + if !fw.committed { + return filepath.Base(fw.tempName) } return filepath.Base(fw.finalName) } func (fw *fileWriter) Close() error { - if fw.File == nil { - return nil + var closeErr, removeErr error + if fw.File != nil { + closeErr = fw.File.Close() + fw.File = nil + } + // Only the first Close of an unpublished writer owns the temp file. A + // second Close must not unlink the path again: another writer may have + // created it in the meantime, and removing it would steal their lock. + if !fw.committed && !fw.aborted { + fw.aborted = true + removeErr = os.Remove(fw.tempName) } - err1 := fw.File.Close() - err2 := os.Remove(fw.File.Name()) - return cmp.Or(err1, err2) + return cmp.Or(closeErr, removeErr) } // fsyncDir flushes the directory entry for path. @@ -71,19 +93,21 @@ func fsyncDir(path string) error { } func (fw *fileWriter) Commit() error { - if err := fw.File.Sync(); err != nil { - return err + if fw.File == nil { + return os.ErrClosed } - if err := fw.File.Close(); err != nil { + if err := fw.File.Sync(); err != nil { return err } - - err := os.Rename(fw.File.Name(), fw.finalName) + err := fw.File.Close() fw.File = nil if err != nil { return err } - + if err := os.Rename(fw.tempName, fw.finalName); err != nil { + return err + } + fw.committed = true return fsyncDir(filepath.Dir(fw.finalName)) } @@ -92,7 +116,7 @@ func newLockForWrite(path string) (AtomicWriter, error) { if err != nil { return nil, err } - return &fileWriter{File: f, finalName: path}, nil + return &fileWriter{File: f, finalName: path, tempName: f.Name()}, nil } func newAtomicWriter(path string) (AtomicWriter, error) { @@ -101,7 +125,7 @@ func newAtomicWriter(path string) (AtomicWriter, error) { if err != nil { return nil, err } - return &fileWriter{File: f, finalName: path}, nil + return &fileWriter{File: f, finalName: path, tempName: f.Name()}, nil } func NewLocalStorage(dir string) *localStorage { diff --git a/storage_test.go b/storage_test.go new file mode 100644 index 0000000..e747693 --- /dev/null +++ b/storage_test.go @@ -0,0 +1,115 @@ +package reftable + +import ( + "os" + "path/filepath" + "strings" + "testing" +) + +func TestFileWriterClosePreservesNewOwnerLock(t *testing.T) { + storage := NewLocalStorage(t.TempDir()) + first, err := storage.LockForWrite(listFileName) + if err != nil { + t.Fatal(err) + } + if err := first.Close(); err != nil { + t.Fatal(err) + } + if first.Committed() { + t.Fatal("aborted writer reports committed") + } + second, err := storage.LockForWrite(listFileName) + if err != nil { + t.Fatal(err) + } + defer second.Close() + if err := first.Close(); err != nil { + t.Fatal(err) + } + third, err := storage.LockForWrite(listFileName) + if err == nil { + third.Close() + t.Fatal("repeated Close removed another writer's lock") + } + if !os.IsExist(err) { + t.Fatalf("got %v, want lock contention", err) + } +} + +func TestFileWriterFailedRenameCanBeCleaned(t *testing.T) { + dir := t.TempDir() + // Replacing a directory with a regular file must fail. + dest := filepath.Join(dir, "destination") + if err := os.Mkdir(dest, 0700); err != nil { + t.Fatal(err) + } + w, err := newAtomicWriter(dest) + if err != nil { + t.Fatal(err) + } + defer w.Close() + temp := filepath.Join(dir, w.Name()) + if err := w.Commit(); err == nil { + t.Fatal("expected rename failure") + } + if w.Committed() { + t.Fatal("failed rename reports committed") + } + if err := w.Close(); err != nil { + t.Fatal(err) + } + if _, err := os.Stat(temp); !os.IsNotExist(err) { + t.Fatalf("temporary file survived cleanup: %v", err) + } +} + +func TestFileWriterCommittedFileSurvivesClose(t *testing.T) { + dir := t.TempDir() + w, err := newAtomicWriter(filepath.Join(dir, "table.ref")) + if err != nil { + t.Fatal(err) + } + defer w.Close() + if _, err := w.Write([]byte("published")); err != nil { + t.Fatal(err) + } + if err := w.Commit(); err != nil { + t.Fatal(err) + } + if !w.Committed() || w.Name() != "table.ref" { + t.Fatal("incorrect state after commit") + } + if err := w.Close(); err != nil { + t.Fatal(err) + } + data, err := os.ReadFile(filepath.Join(dir, "table.ref")) + if err != nil || string(data) != "published" { + t.Fatalf("published file changed after Close: %q, %v", data, err) + } +} + +// Name() must not start pointing at the final path once an unpublished writer +// is closed: Remove(w.Name()) would then delete the live file. +func TestFileWriterNameStableAcrossAbortedClose(t *testing.T) { + dir := t.TempDir() + s := NewLocalStorage(dir) + + w, err := s.LockForWrite("tables.list") + if err != nil { + t.Fatal(err) + } + before := w.Name() + if err := w.Close(); err != nil { + t.Fatal(err) + } + if after := w.Name(); after != before { + t.Errorf("Name() changed across an aborted Close: %q -> %q", before, after) + } + if w.Committed() { + t.Error("an aborted writer reports Committed") + } + if !strings.Contains(before, "tables.list") { + t.Errorf("unexpected lock name %q", before) + } +} diff --git a/validation_test.go b/validation_test.go new file mode 100644 index 0000000..8bc9ca6 --- /dev/null +++ b/validation_test.go @@ -0,0 +1,153 @@ +// Tests that malformed or hostile input is rejected rather than panicking, +// over-allocating, or escaping the reftable directory. + +package reftable + +import ( + "bytes" + "errors" + "fmt" + "os" + "path/filepath" + "testing" +) + +// A manifest entry is joined onto the reftable directory, and filepath.Join +// cleans "..", so an unvalidated name addresses arbitrary paths. +func TestRejectsTraversalInTablesList(t *testing.T) { + for _, name := range []string{ + "../escape.ref", + "../../../etc/passwd", + "sub/dir.ref", + `sub\dir.ref`, + "..", + ".", + "", + } { + if err := validateTableName(name); err == nil { + t.Errorf("validateTableName(%q) = nil, want error", name) + } + } + if err := validateTableName("0x000000000001-0x000000000001-abcdef12.ref"); err != nil { + t.Errorf("validateTableName(valid) = %v, want nil", err) + } +} + +// End to end: a hostile tables.list must not open, or later remove, a table +// outside the reftable directory. +func TestStackRejectsTraversalManifest(t *testing.T) { + root := t.TempDir() + rtDir := filepath.Join(root, "reftable") + outside := filepath.Join(root, "outside") + for _, d := range []string{rtDir, outside} { + if err := os.MkdirAll(d, 0o755); err != nil { + t.Fatal(err) + } + } + + victim := filepath.Join(outside, "victim.ref") + _, _ = constructTestTable(t, []RefRecord{{RefName: "refs/heads/v", UpdateIndex: 1, Value: testHash(1)}}, nil, Config{}) + if err := os.WriteFile(victim, []byte("not a reftable"), 0o644); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(rtDir, "tables.list"), + []byte("../outside/victim.ref\n"), 0o644); err != nil { + t.Fatal(err) + } + + st, err := NewStack(NewLocalStorage(rtDir), Config{}) + if err == nil { + st.Close() + t.Fatal("NewStack accepted a manifest entry outside the reftable dir") + } + if !errors.Is(err, fmtError) { + t.Errorf("got %v, want a format error", err) + } + if _, err := os.Stat(victim); err != nil { + t.Errorf("file outside the reftable dir was disturbed: %v", err) + } +} + +// The hash ID is 4 raw bytes from the header and is consumed before the CRC is +// compared, so an unknown value must be an error rather than a panic. +func TestUnknownHashIDIsAnError(t *testing.T) { + // SHA-256 forces a v2 header, which carries the hash id explicitly. + _, reader := constructTestTable(t, []RefRecord{ + {RefName: "refs/heads/main", UpdateIndex: 1, Value: make([]byte, 32)}, + }, nil, Config{HashID: SHA256ID}) + + raw := append([]byte(nil), reader.src.(*ByteBlockSource).Source...) + if n := bytes.Count(raw, []byte("s256")); n != 2 { + t.Fatalf("expected 2 copies of the hash id in the header/footer, got %d", n) + } + // Corrupt both copies so the start/tail header comparison still passes and + // the unknown hash id is what NewReader actually trips over. The CRC is + // left stale on purpose: this must fail before the CRC is even checked. + raw = bytes.ReplaceAll(raw, []byte("s256"), []byte("XXXX")) + + if _, err := NewReader(&ByteBlockSource{Source: raw}, "corrupt"); err == nil { + t.Fatal("NewReader accepted an unknown hash id") + } +} + +func TestNewWriterRejectsBadConfig(t *testing.T) { + if _, err := NewWriter(&bytes.Buffer{}, &Config{HashID: HashID{'n', 'o', 'p', 'e'}}); err == nil { + t.Error("NewWriter accepted an unknown hash id") + } + // Must be rejected without first allocating a 1GiB block. + if _, err := NewWriter(&bytes.Buffer{}, &Config{BlockSize: 1 << 30}); err == nil { + t.Error("NewWriter accepted an oversized block size") + } +} + +func TestNewWriterRejectsSmallBlocks(t *testing.T) { + for _, hashID := range []HashID{NullHashID, SHA1ID, SHA256ID} { + version := 1 + if hashID == SHA256ID { + version = 2 + } + // The file header, block header, and restart count must all fit. + minimum := uint32(headerSize(version) + 4 + 2) + for size := uint32(1); size < minimum; size++ { + t.Run(fmt.Sprintf("hash=%x/size=%d", hashID, size), func(t *testing.T) { + if _, err := NewWriter(&bytes.Buffer{}, &Config{HashID: hashID, BlockSize: size}); err == nil { + t.Fatal("NewWriter accepted a block too small for its headers") + } + }) + } + for _, size := range []uint32{0, minimum} { + t.Run(fmt.Sprintf("hash=%x/valid-size=%d", hashID, size), func(t *testing.T) { + w, err := NewWriter(&bytes.Buffer{}, &Config{HashID: hashID, BlockSize: size}) + if err != nil { + t.Fatal(err) + } + w.SetLimits(1, 1) + err = w.AddRef(&RefRecord{RefName: "refs/heads/a", UpdateIndex: 1, Value: make([]byte, hashID.Size())}) + if size == minimum { + // Structurally large enough for headers, but not this record. + if err == nil { + t.Fatal("expected record-too-large error") + } + } else if err != nil { + t.Fatal(err) + } + if err := w.Close(); err != nil && !errors.Is(err, ErrEmptyTable) { + t.Fatal(err) + } + }) + } + } +} + +// A log block declares its decompressed size. DEFLATE cannot expand by more +// than 1032:1, so a huge declared size over a tiny block is malformed and must +// not be preallocated. +func TestLogBlockRejectsImpossibleSize(t *testing.T) { + block := make([]byte, 64) + block[0] = blockTypeLog + block[1], block[2], block[3] = 0xff, 0xff, 0xff // sz = 16MiB from a 64 byte block + + if _, err := newBlockReader(block, 0, 64, 20); err == nil { + t.Fatal("newBlockReader accepted a log block declaring an impossible size") + } +} diff --git a/writer.go b/writer.go index e9c9658..13381e4 100644 --- a/writer.go +++ b/writer.go @@ -86,15 +86,31 @@ func (cfg *Config) setDefaults() { func NewWriter(out io.Writer, cfg *Config) (*Writer, error) { o := *cfg o.setDefaults() + + // Validate before allocating: BlockSize is caller-supplied, so checking + // after make() lets Config{BlockSize: 1<<30} reserve 1GiB and then fail. + if o.BlockSize >= (1 << 24) { + return nil, errors.New("reftable: invalid blocksize") + } + if o.HashID.Size() == 0 { + return nil, fmt.Errorf("reftable: unknown hash id %q", string(o.HashID[:])) + } + version := 1 + if o.HashID == SHA256ID { + version = 2 + } + // Reserve the file header, the four-byte block header, and the + // two-byte restart count before initializing the first block writer. + minimumBlockSize := uint32(headerSize(version) + 4 + 2) + if o.BlockSize < minimumBlockSize { + return nil, fmt.Errorf("reftable: block size %d is smaller than minimum %d", o.BlockSize, minimumBlockSize) + } + w := &Writer{ cfg: o, block: make([]byte, o.BlockSize), } - if cfg.BlockSize >= (1 << 24) { - return nil, errors.New("reftable: invalid blocksize") - } - w.paddedWriter.out = out if !cfg.SkipIndexObjects { w.objIndex = map[string][]uint64{} @@ -485,12 +501,17 @@ func (w *Writer) finishSection() error { threshold = 1 } before := w.Stats.idxStats.Blocks + // Build index levels bottom-up. Each flushBlock appends an index record + // for the block it wrote, so w.index accumulates the next level up as we + // go. The loop stops once the top level is small enough for the reader to + // scan linearly (seekLinear walks across blocks), which is what threshold + // expresses. for len(w.index) > threshold { maxLevel++ indexStart = w.next - w.blockWriter = w.newBlockWriter(blockTypeIndex) idx := w.index w.index = nil + w.blockWriter = w.newBlockWriter(blockTypeIndex) for _, i := range idx { if w.blockWriter.add(&i) { continue @@ -504,11 +525,34 @@ func (w *Writer) finishSection() error { panic("fail on fresh block") } } + + // Flush this level's final, partial block. Without this the + // pending block is either silently discarded by the next + // iteration (losing every entry in it) or flushed after w.index + // has been cleared, leaving a stale record that is then written + // as the first entry of the *next* section's index. + if err := w.flushBlock(); err != nil { + return err + } + + if len(w.index) >= len(idx) { + // The level did not shrink, so no further level can collapse + // it either: the keys are large enough that an index block + // holds a single entry. Stop instead of looping forever. + // See TestTableObjectIDLen for this shape. + break + } } + + // No flush is needed here. Every path above leaves w.blockWriter either + // nil or holding zero entries, and flushBlock returns early for both: + // each loop iteration ends by flushing its level, and the flush at the + // top of this function already handled the section's last data block. + // + // Drop the index. Its remaining records describe this section's own + // top-level index blocks; carrying them into the next section would + // write them as that section's first index entries. w.index = nil - if err := w.flushBlock(); err != nil { - return err - } blockStats := w.getBlockStats(typ) blockStats.IndexBlocks = w.Stats.idxStats.Blocks - before diff --git a/writer_index_test.go b/writer_index_test.go new file mode 100644 index 0000000..deb2bbc --- /dev/null +++ b/writer_index_test.go @@ -0,0 +1,51 @@ +package reftable + +import ( + "fmt" + "testing" +) + +// finishSection must flush every index level's final block and must not carry +// index records into the next section. Previously this shape lost reflogs +// through the index with no error reported. +func TestIndexCoversEveryRecord(t *testing.T) { + for _, blockSize := range []uint32{256, 512, 4096} { + t.Run(fmt.Sprintf("blockSize=%d", blockSize), func(t *testing.T) { + var refs []RefRecord + var logs []LogRecord + for i := range 500 { + name := fmt.Sprintf("refs/heads/b%06d", i) + refs = append(refs, RefRecord{RefName: name, UpdateIndex: 1, Value: testHash(i)}) + logs = append(logs, LogRecord{ + RefName: name, UpdateIndex: 1, + New: testHash(i), Old: testHash(i), + Name: "n", Email: "e@x", Message: "m", + }) + } + + writer, reader := constructTestTable(t, refs, logs, Config{BlockSize: blockSize}) + if n := len(writer.index); n != 0 { + t.Errorf("writer kept %d index record(s) after the last section", n) + } + + for _, r := range refs { + got, err := ReadRef(reader, r.RefName) + if err != nil { + t.Fatalf("ReadRef(%s): %v", r.RefName, err) + } + if got == nil { + t.Fatalf("ref %s unreachable through the index", r.RefName) + } + } + for _, l := range logs { + got, err := ReadLogAt(reader, l.RefName, 1) + if err != nil { + t.Fatalf("ReadLogAt(%s): %v", l.RefName, err) + } + if got == nil { + t.Fatalf("log %s unreachable through the index", l.RefName) + } + } + }) + } +}