slicer: Do not drop link object's payload (#560)

object/slicer/slicer.go

@@ -573,13 +573,18 @@ func (x *PayloadWriter) _writeChild(ctx context.Context, meta dynamicObjectMetad
 		linkObj.SetObjects(x.writtenChildren)
 		obj.WriteLink(linkObj)
 
-		meta.reset()
 		obj.ResetPreviousID()
 		// we reuse already written object, we should reset these fields, to eval them one more time in writeInMemObject.
 		obj.ResetID()
 		obj.SetSignature(nil)
 
-		_, err = writeInMemObject(ctx, x.signer, x.stream, obj, nil, meta, x.prmObjectPutInit)
+		payload := obj.Payload()
+		payloadAsBuffers := [][]byte{obj.Payload()}
+
+		meta.reset()
+		meta.accumulateNextPayloadChunk(payload)
+
+		_, err = writeInMemObject(ctx, x.signer, x.stream, obj, payloadAsBuffers, meta, x.prmObjectPutInit)
 		if err != nil {
 			return fmt.Errorf("write linking object: %w", err)
 		}

object/slicer/slicer_test.go

@@ -479,15 +479,14 @@ func (x *slicedObjectChecker) ObjectPutInit(_ context.Context, hdr object.Object
 }
 
 type writeSizeChecker struct {
-	tb          testing.TB
-	hdr         object.Object
-	limit       uint64
-	processed   uint64
-	base        io.Writer
-	payloadSeen bool
+	tb        testing.TB
+	hdr       object.Object
+	limit     uint64
+	processed uint64
+	base      *bytes.Buffer
 }
 
-func newSizeChecker(tb testing.TB, hdr object.Object, base io.Writer, sizeLimit uint64) *writeSizeChecker {
+func newSizeChecker(tb testing.TB, hdr object.Object, base *bytes.Buffer, sizeLimit uint64) *writeSizeChecker {
 	return &writeSizeChecker{
 		tb:    tb,
 		hdr:   hdr,
@@ -497,28 +496,29 @@ func newSizeChecker(tb testing.TB, hdr object.Object, base io.Writer, sizeLimit
 }
 
 func (x *writeSizeChecker) Write(p []byte) (int, error) {
-	if !x.payloadSeen && len(p) > 0 {
-		x.payloadSeen = true
-	}
-
-	if x.payloadSeen {
-		if len(x.hdr.Children()) == 0 {
-			// only linking objects should be streamed with
-			// empty payload
-			require.NotZero(x.tb, len(p))
-		} else {
-			// linking object should have empty payload
-			require.Zero(x.tb, x.hdr.PayloadSize())
-		}
-	}
+	require.NotZero(x.tb, len(p), "non of the split object should be empty")
 
 	n, err := x.base.Write(p)
 	x.processed += uint64(n)
 	return n, err
 }
 
 func (x *writeSizeChecker) Close() error {
-	require.LessOrEqual(x.tb, x.processed, x.limit, "object payload must not overflow the limit")
+	if x.hdr.Type() == object.TypeLink {
+		payload := x.base.Bytes()
+
+		var testLink object.Link
+		require.NoError(x.tb, testLink.Unmarshal(payload), "link object's payload must be structured")
+	} else {
+		require.LessOrEqual(x.tb, x.processed, x.limit, "object payload must not overflow the limit")
+	}
+
+	require.Equal(x.tb, x.processed, x.hdr.PayloadSize())
+
+	// deprecated things
+	require.Nil(x.tb, x.hdr.SplitID(), "no split ID should be presented")
+	require.Empty(x.tb, x.hdr.Children(), "no child should be stored in the headers")
+
 	return nil
 }