filer: let the upload loop cut chunks at caller-chosen boundaries

The chunking loop always cut at a fixed size. Accept a ChunkBoundaries
source instead, with the fixed size as the default implementation, so a
caller can align storage chunks to structure inside the file. Inline
small-content storage is disabled in that mode because it would drop the
first boundary.

weed/server/filer_server_handlers_write_upload.go

@@ -29,6 +29,17 @@ var bufPool = sync.Pool{
 	},
 }
 
+// ChunkBoundaries decides where the upload loop may cut a storage chunk.
+type ChunkBoundaries interface {
+	// NextChunkSize returns the size of the chunk starting at offset, or 0 when
+	// no further chunks are expected.
+	NextChunkSize(offset int64) int64
+}
+
+type fixedChunkSize int64
+
+func (c fixedChunkSize) NextChunkSize(int64) int64 { return int64(c) }
+
 func (fs *FilerServer) uploadRequestToChunks(ctx context.Context, w http.ResponseWriter, r *http.Request, reader io.Reader, chunkSize int32, fileName, contentType string, contentLength int64, so *operation.StorageOption) (fileChunks []*filer_pb.FileChunk, md5Hash hash.Hash, chunkOffset int64, uploadErr error, smallContent []byte) {
 	query := r.URL.Query()
 
@@ -51,6 +62,13 @@ func (fs *FilerServer) uploadRequestToChunks(ctx context.Context, w http.Respons
 }
 
 func (fs *FilerServer) uploadReaderToChunks(ctx context.Context, r *http.Request, reader io.Reader, startOffset int64, chunkSize int32, fileName, contentType string, isAppend bool, so *operation.StorageOption) (fileChunks []*filer_pb.FileChunk, md5Hash hash.Hash, chunkOffset int64, uploadErr error, smallContent []byte) {
+	return fs.uploadReaderToBoundedChunks(ctx, r, reader, startOffset, fixedChunkSize(chunkSize), true, fileName, contentType, isAppend, so)
+}
+
+// uploadReaderToBoundedChunks cuts chunks where boundaries allows instead of at
+// a fixed size. allowInline permits the small-content optimization, which must
+// stay off when chunk boundaries carry meaning.
+func (fs *FilerServer) uploadReaderToBoundedChunks(ctx context.Context, r *http.Request, reader io.Reader, startOffset int64, boundaries ChunkBoundaries, allowInline bool, fileName, contentType string, isAppend bool, so *operation.StorageOption) (fileChunks []*filer_pb.FileChunk, md5Hash hash.Hash, chunkOffset int64, uploadErr error, smallContent []byte) {
 
 	md5Hash = md5.New()
 	chunkOffset = startOffset
@@ -63,6 +81,11 @@ func (fs *FilerServer) uploadReaderToChunks(ctx context.Context, r *http.Request
 	var uploadErrLock sync.Mutex
 	for {
 
+		wantSize := boundaries.NextChunkSize(chunkOffset)
+		if wantSize <= 0 {
+			break
+		}
+
 		// need to throttle used byte buffer
 		bytesBufferLimitChan <- struct{}{}
 
@@ -78,7 +101,7 @@ func (fs *FilerServer) uploadReaderToChunks(ctx context.Context, r *http.Request
 
 		bytesBuffer := bufPool.Get().(*bytes.Buffer)
 
-		limitedReader := io.LimitReader(partReader, int64(chunkSize))
+		limitedReader := io.LimitReader(partReader, wantSize)
 
 		bytesBuffer.Reset()
 
@@ -97,7 +120,7 @@ func (fs *FilerServer) uploadReaderToChunks(ctx context.Context, r *http.Request
 			}
 			break
 		}
-		if chunkOffset == 0 && !isAppend {
+		if chunkOffset == 0 && !isAppend && allowInline {
 			if dataSize < fs.option.SaveToFilerLimit {
 				chunkOffset += dataSize
 				smallContent = make([]byte, dataSize)
@@ -141,7 +164,7 @@ func (fs *FilerServer) uploadReaderToChunks(ctx context.Context, r *http.Request
 		chunkOffset = chunkOffset + dataSize
 
 		// if last chunk was not at full chunk size, but already exhausted the reader
-		if dataSize < int64(chunkSize) {
+		if dataSize < wantSize {
 			break
 		}
 	}