From 79fd0b19cfa5a0d47bff3b97d02f767257cbb8ad Mon Sep 17 00:00:00 2001 From: Kovid Goyal Date: Mon, 3 Jul 2023 17:47:33 +0530 Subject: [PATCH] Fix a bug in upstream CreateDelta that could ignore trailing data if a matching hash is not found for it --- tools/rsync/algorithm.go | 27 ++++++++++++++++++--------- tools/rsync/api_test.go | 27 ++++++++++++++++----------- 2 files changed, 34 insertions(+), 20 deletions(-) diff --git a/tools/rsync/algorithm.go b/tools/rsync/algorithm.go index 3406c1372..539f3cfa7 100644 --- a/tools/rsync/algorithm.go +++ b/tools/rsync/algorithm.go @@ -267,11 +267,10 @@ func (r *RSync) CreateDelta(source io.Reader, signature []BlockHash, ops Operati // Send the last operation if there is one waiting. defer func() { - if prevOp == nil { - return + if prevOp != nil { + err = ops(*prevOp) + prevOp = nil } - err = ops(*prevOp) - prevOp = nil }() // Combine OpBlock into OpBlockRange. To do this store the previous @@ -320,6 +319,14 @@ func (r *RSync) CreateDelta(source io.Reader, signature []BlockHash, ops Operati return } + send_data := func() error { + if err := enqueue(Operation{Type: OpData, Data: buffer[data.tail:data.head]}); err != nil { + return err + } + data.tail = data.head + return nil + } + for !lastRun { // Determine if the buffer should be extended. if sum.tail+r.BlockSize > validTo { @@ -327,8 +334,7 @@ func (r *RSync) CreateDelta(source io.Reader, signature []BlockHash, ops Operati if validTo+r.BlockSize > len(buffer) { // Before wrapping the buffer, send any trailing data off. if data.tail < data.head { - err = enqueue(Operation{Type: OpData, Data: buffer[data.tail:data.head]}) - if err != nil { + if err = send_data(); err != nil { return err } } @@ -354,6 +360,11 @@ func (r *RSync) CreateDelta(source io.Reader, signature []BlockHash, ops Operati data.head = validTo } if n == 0 { + if data.tail < data.head { + if err = send_data(); err != nil { + return err + } + } break } } @@ -382,11 +393,9 @@ func (r *RSync) CreateDelta(source io.Reader, signature []BlockHash, ops Operati // must be flushed first), or the data chunk size has reached it's maximum size (for buffer // allocation purposes) or to flush the end of the data. if data.tail < data.head && (foundHash || data.head-data.tail >= r.MaxDataOp || lastRun) { - err = enqueue(Operation{Type: OpData, Data: buffer[data.tail:data.head]}) - if err != nil { + if err = send_data(); err != nil { return err } - data.tail = data.head } if foundHash { diff --git a/tools/rsync/api_test.go b/tools/rsync/api_test.go index 54f43eb60..b8bf8f034 100644 --- a/tools/rsync/api_test.go +++ b/tools/rsync/api_test.go @@ -54,17 +54,22 @@ func TestRsyncRoundtrip(t *testing.T) { signature = append(signature, s) return nil }) - delta_ops := make([]Operation, 0, 128) - p.rsync.CreateDelta(bytes.NewReader(src_data), signature, func(op Operation) error { - op.Data = slices.Clone(op.Data) - delta_ops = append(delta_ops, op) - return nil - }) - outputbuf := bytes.Buffer{} - for _, op := range delta_ops { - p.rsync.ApplyDelta(&outputbuf, bytes.NewReader(src_data), op) + + apply_delta := func(signature []BlockHash) []byte { + delta_ops := make([]Operation, 0, 1024) + p.rsync.CreateDelta(bytes.NewReader(src_data), signature, func(op Operation) error { + op.Data = slices.Clone(op.Data) + delta_ops = append(delta_ops, op) + return nil + }) + outputbuf := bytes.Buffer{} + for _, op := range delta_ops { + p.rsync.ApplyDelta(&outputbuf, bytes.NewReader(src_data), op) + } + return outputbuf.Bytes() } - test_equal(src_data, outputbuf.Bytes()) + test_equal(src_data, apply_delta(nil)) + test_equal(src_data, apply_delta(signature)) // Now try with serialization p = NewPatcher(int64(len(src_data))) @@ -80,7 +85,7 @@ func TestRsyncRoundtrip(t *testing.T) { if err := d.CreateDelta(bytes.NewReader(src_data), func(b []byte) error { _, err := deltabuf.Write(b); return err }); err != nil { t.Fatal(err) } - outputbuf = bytes.Buffer{} + outputbuf := bytes.Buffer{} p.StartDelta(&outputbuf, bytes.NewReader(src_data)) b := make([]byte, 30*1024) for {