diff --git a/kittens/transfer/algorithm.c b/kittens/transfer/algorithm.c index e0ef72a72..26bacada8 100644 --- a/kittens/transfer/algorithm.c +++ b/kittens/transfer/algorithm.c @@ -190,6 +190,7 @@ typedef struct { Rsync rsync; buffer buf, block_buf; PyObject *block_buf_view; + bool checksum_done; } Patcher; static int @@ -350,7 +351,9 @@ apply_op(Patcher *self, Operation op, PyObject *read, PyObject *write) { DECREF_AFTER_FUNCTION PyObject *b2 = PyBytes_FromStringAndSize((char*)op.data.buf, self->rsync.checksummer.hash_size); DECREF_AFTER_FUNCTION PyObject *h2 = PyObject_CallMethod(b2, "hex", NULL); PyErr_Format(RsyncError, "Failed to verify overall file checksum actual: %S != expected: %S, this usually happens because one of the involved files was altered while the operation was in progress.", h1, h2); - return false; } + return false; + } + self->checksum_done = true; } return true; } PyErr_SetString(RsyncError, "Unknown operation type"); @@ -379,6 +382,7 @@ apply_delta_data(Patcher *self, PyObject *args) { static PyObject* finish_delta_data(Patcher *self, PyObject *args UNUSED) { if (self->buf.len > 0) { PyErr_Format(RsyncError, "%zu bytes of unused delta data", self->buf.len); return NULL; } + if (!self->checksum_done) { PyErr_SetString(RsyncError, "The checksum was not received at the end of the delta data"); return NULL; } Py_RETURN_NONE; } diff --git a/kittens/transfer/receive.go b/kittens/transfer/receive.go index d12eabded..ab15906d3 100644 --- a/kittens/transfer/receive.go +++ b/kittens/transfer/receive.go @@ -62,6 +62,62 @@ func (ff *filesystem_file) write(data []byte) (int, error) { return n, err } +type patch_file struct { + path string + src, temp *os.File + p *rsync.Patcher +} + +func (pf *patch_file) tell() (int64, error) { + if pf.temp == nil { + s, err := os.Stat(pf.path) + return s.Size(), err + } + return pf.temp.Seek(0, os.SEEK_CUR) +} + +func (pf *patch_file) close() (err error) { + if pf.p == nil { + return + } + err = pf.p.FinishDelta() + pf.src.Close() + pf.temp.Close() + if err == nil { + err = os.Rename(pf.temp.Name(), pf.src.Name()) + } + pf.src = nil + pf.temp = nil + pf.p = nil + return +} + +func (pf *patch_file) write(data []byte) (int, error) { + if err := pf.p.UpdateDelta(data); err == nil { + return len(data), nil + } else { + return 0, err + } +} + +func new_patch_file(path string, p *rsync.Patcher) (ans *patch_file, err error) { + ans = &patch_file{p: p, path: path} + var f *os.File + if f, err = os.Open(path); err != nil { + return + } else { + ans.src = f + } + if f, err = os.CreateTemp(filepath.Dir(path), ""); err != nil { + ans.src.Close() + return + } else { + ans.temp = f + } + ans.p.StartDelta(ans.temp, ans.src) + return +} + type remote_file struct { expected_size int64 expect_diff bool @@ -84,6 +140,7 @@ type remote_file struct { compression_type Compression remote_symlink_value string actual_file output_file + patch_file patch_file } func (self *remote_file) close() error { @@ -109,7 +166,11 @@ func (self *remote_file) write_data(data []byte, is_last bool) (amt_written int6 os.MkdirAll(parent, 0o755) } if self.expect_diff { - panic(`TODO: create PatchFile for rsync`) + if pf, err := new_patch_file(self.expanded_local_path, self.patcher); err != nil { + return err + } else { + self.actual_file = pf + } } else { if ff, err := os.Create(self.expanded_local_path); err != nil { return err diff --git a/kitty/file_transmission.py b/kitty/file_transmission.py index fbb6dd8fa..6aa38d38f 100644 --- a/kitty/file_transmission.py +++ b/kitty/file_transmission.py @@ -396,13 +396,13 @@ class PatchFile: self.closed = True p = self.patcher del self.block_buffer, self.patcher - if self.src_file is not None and not self.src_file.closed: - self.src_file.close() if self._dest_file is not None and not self._dest_file.closed: self._dest_file.close() p.finish_delta_data() if self.src_file is not None: os.replace(self.dest_file.name, self.src_file.name) + if self.src_file is not None and not self.src_file.closed: + self.src_file.close() def tell(self) -> int: df = self.dest_file diff --git a/tools/rsync/algorithm.go b/tools/rsync/algorithm.go index 417950edc..8d5953a11 100644 --- a/tools/rsync/algorithm.go +++ b/tools/rsync/algorithm.go @@ -195,6 +195,7 @@ type rsync struct { hasher_constructor func() hash.Hash64 checksummer_constructor func() hash.Hash checksummer hash.Hash + checksum_done bool buffer []byte } @@ -319,6 +320,7 @@ func (r *rsync) ApplyDelta(alignedTarget io.Writer, target io.ReadSeeker, op Ope if !bytes.Equal(actual, op.Data) { return fmt.Errorf("Failed to verify overall file checksum actual: %s != expected: %s. This usually happens if some data was corrupted in transit or one of the involved files was altered while the transfer was in progress.", hex.EncodeToString(actual), hex.EncodeToString(op.Data)) } + r.checksum_done = true } return nil } diff --git a/tools/rsync/api.go b/tools/rsync/api.go index 23f185a54..4df743e12 100644 --- a/tools/rsync/api.go +++ b/tools/rsync/api.go @@ -185,6 +185,9 @@ func (self *Patcher) FinishDelta() (err error) { self.delta_input = nil self.delta_output = nil self.unconsumed_delta_data = nil + if !self.rsync.checksum_done { + return fmt.Errorf("The checksum was not received at the end of the delta data") + } return }