More work on porting the receive kitten

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;
 }
 

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

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

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
 }

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
 }