Merge pull request #13 from SiaFoundation/single-slab

Single-slab endpoints

Author: lukechampine

api/api.go

@@ -198,10 +198,8 @@ type SlabsUploadRequest struct {
 
 // SlabsDownloadRequest is the request type for the /slabs/download endpoint.
 type SlabsDownloadRequest struct {
-	Slabs     []slab.Slice `json:"slabs"`
-	Offset    int64        `json:"offset"`
-	Length    int64        `json:"length"`
-	Contracts []Contract   `json:"contracts"`
+	Slab      slab.Slice `json:"slab"`
+	Contracts []Contract `json:"contracts"`
 }
 
 // SlabsDeleteRequest is the request type for the /slabs/delete endpoint.
@@ -212,10 +210,10 @@ type SlabsDeleteRequest struct {
 
 // SlabsMigrateRequest is the request type for the /slabs/migrate endpoint.
 type SlabsMigrateRequest struct {
-	Slabs         []slab.Slab `json:"slabs"`
-	From          []Contract  `json:"from"`
-	To            []Contract  `json:"to"`
-	CurrentHeight uint64      `json:"currentHeight"`
+	Slab          slab.Slab  `json:"slab"`
+	From          []Contract `json:"from"`
+	To            []Contract `json:"to"`
+	CurrentHeight uint64     `json:"currentHeight"`
 }
 
 // ObjectsResponse is the response type for the /objects endpoint.

api/api_test.go

@@ -95,20 +95,20 @@ type mockSlabMover struct {
 	hosts []slab.Host
 }
 
-func (sm *mockSlabMover) UploadSlabs(ctx context.Context, r io.Reader, m, n uint8, currentHeight uint64, contracts []api.Contract) ([]slab.Slab, error) {
-	return slab.UploadSlabs(r, m, n, sm.hosts)
+func (sm *mockSlabMover) UploadSlab(ctx context.Context, r io.Reader, m, n uint8, currentHeight uint64, contracts []api.Contract) (slab.Slab, error) {
+	return slab.UploadSlab(r, m, n, sm.hosts)
 }
 
-func (sm *mockSlabMover) DownloadSlabs(ctx context.Context, w io.Writer, slabs []slab.Slice, offset, length int64, contracts []api.Contract) error {
-	return slab.DownloadSlabs(w, slabs, offset, length, sm.hosts)
+func (sm *mockSlabMover) DownloadSlab(ctx context.Context, w io.Writer, s slab.Slice, contracts []api.Contract) error {
+	return slab.DownloadSlab(w, s, sm.hosts)
 }
 
 func (sm *mockSlabMover) DeleteSlabs(ctx context.Context, slabs []slab.Slab, contracts []api.Contract) error {
 	return slab.DeleteSlabs(slabs, sm.hosts)
 }
 
-func (sm *mockSlabMover) MigrateSlabs(ctx context.Context, slabs []slab.Slab, currentHeight uint64, from, to []api.Contract) error {
-	return slab.MigrateSlabs(slabs, sm.hosts, sm.hosts)
+func (sm *mockSlabMover) MigrateSlab(ctx context.Context, s *slab.Slab, currentHeight uint64, from, to []api.Contract) error {
+	return slab.MigrateSlab(s, sm.hosts, sm.hosts)
 }
 
 type node struct {
@@ -144,7 +144,7 @@ func runServer(n *node) (*api.Client, func()) {
 	}
 	go func() {
 		srv := api.NewServer(mockSyncer{}, mockChainManager{}, mockTxPool{}, n.w, n.hdb, mockRHP{}, n.cs, n.sm, n.os)
-		http.Serve(l, jape.AuthMiddleware(srv, "password"))
+		http.Serve(l, jape.BasicAuth("password")(srv))
 	}()
 	c := api.NewClient("http://"+l.Addr().String(), "password")
 	return c, func() { l.Close() }
@@ -199,20 +199,17 @@ func TestObject(t *testing.T) {
 	// upload
 	data := frand.Bytes(12345)
 	key := object.GenerateEncryptionKey()
-	slabs, err := c.UploadSlabs(key.Encrypt(bytes.NewReader(data)), 2, 3, 0, contracts)
+	s, err := c.UploadSlab(key.Encrypt(bytes.NewReader(data)), 2, 3, 0, contracts)
 	if err != nil {
 		t.Fatal(err)
 	}
 	o := object.Object{
-		Key:   key,
-		Slabs: make([]slab.Slice, len(slabs)),
-	}
-	for i := range slabs {
-		o.Slabs[i] = slab.Slice{
-			Slab:   slabs[i],
+		Key: key,
+		Slabs: []slab.Slice{{
+			Slab:   s,
 			Offset: 0,
 			Length: uint32(len(data)),
-		}
+		}},
 	}
 
 	// store object
@@ -228,17 +225,17 @@ func TestObject(t *testing.T) {
 
 	// download
 	var buf bytes.Buffer
-	if err := c.DownloadSlabs(key.Decrypt(&buf, 0), o.Slabs, 0, o.Size(), contracts); err != nil {
+	if err := c.DownloadSlab(key.Decrypt(&buf, 0), o.Slabs[0], contracts); err != nil {
 		t.Fatal(err)
 	} else if !bytes.Equal(buf.Bytes(), data) {
 		t.Fatalf("data mismatch:\n%v (%v)\n%v (%v)", buf.Bytes(), len(buf.Bytes()), data, len(data))
 	}
 
 	// delete slabs
-	if err := c.DeleteSlabs(slabs, contracts); err != nil {
+	if err := c.DeleteSlabs([]slab.Slab{s}, contracts); err != nil {
 		t.Fatal(err)
 	}
-	if err := c.DownloadSlabs(ioutil.Discard, o.Slabs, 0, o.Size(), contracts); err == nil {
+	if err := c.DownloadSlab(ioutil.Discard, o.Slabs[0], contracts); err == nil {
 		t.Error("slabs should no longer be retrievable")
 	}
 

api/client.go

@@ -7,7 +7,6 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
-	"mime/multipart"
 	"net/http"
 	"time"
 
@@ -348,58 +347,43 @@ func (c *Client) HostSetResolves(name string) (ips []string, err error) {
 	return
 }
 
-// UploadSlabs uploads data to a set of hosts.
-func (c *Client) UploadSlabs(src io.Reader, m, n uint8, height uint64, contracts []Contract) (slabs []slab.Slab, err error) {
-	c.c.Custom("POST", "/slabs/upload", []byte{}, &slabs)
-
-	// apparently, the only way to stream a multipart upload is via io.Pipe :/
-	r, w := io.Pipe()
-	mw := multipart.NewWriter(w)
-	errChan := make(chan error, 1)
-	go func() {
-		defer w.Close()
-		defer mw.Close()
-		js, _ := json.Marshal(SlabsUploadRequest{
-			MinShards:     m,
-			TotalShards:   n,
-			Contracts:     contracts,
-			CurrentHeight: height,
-		})
-		mw.WriteField("meta", string(js))
-		part, _ := mw.CreateFormFile("data", "data")
-		_, err := io.Copy(part, src)
-		errChan <- err
-	}()
-	req, err := http.NewRequest("POST", fmt.Sprintf("%v%v", c.c.BaseURL, "/slabs/upload"), r)
+// UploadSlab uploads data to a set of hosts. At most m*SectorSize bytes will be
+// read from src.
+func (c *Client) UploadSlab(src io.Reader, m, n uint8, height uint64, contracts []Contract) (s slab.Slab, err error) {
+	c.c.Custom("POST", "/slabs/upload", []byte{}, &s)
+
+	js, _ := json.Marshal(SlabsUploadRequest{
+		MinShards:     m,
+		TotalShards:   n,
+		Contracts:     contracts,
+		CurrentHeight: height,
+	})
+	body := io.MultiReader(bytes.NewReader(js), io.LimitReader(src, int64(m)*rhpv2.SectorSize))
+	req, err := http.NewRequest("POST", fmt.Sprintf("%v%v", c.c.BaseURL, "/slabs/upload"), body)
 	if err != nil {
 		panic(err)
 	}
-	req.Header.Set("Content-Type", mw.FormDataContentType())
 	req.SetBasicAuth("", c.c.Password)
 	resp, err := http.DefaultClient.Do(req)
 	if err != nil {
-		return nil, err
-	} else if err := <-errChan; err != nil {
-		return nil, err
+		return slab.Slab{}, err
 	}
 	defer io.Copy(ioutil.Discard, resp.Body)
 	defer resp.Body.Close()
 	if resp.StatusCode != 200 {
 		err, _ := ioutil.ReadAll(resp.Body)
-		return nil, errors.New(string(err))
+		return slab.Slab{}, errors.New(string(err))
 	}
-	err = json.NewDecoder(resp.Body).Decode(&slabs)
+	err = json.NewDecoder(resp.Body).Decode(&s)
 	return
 }
 
-// DownloadSlabs downloads data from a set of hosts.
-func (c *Client) DownloadSlabs(dst io.Writer, slabs []slab.Slice, offset, length int64, contracts []Contract) (err error) {
+// DownloadSlab downloads data from a set of hosts.
+func (c *Client) DownloadSlab(dst io.Writer, s slab.Slice, contracts []Contract) (err error) {
 	c.c.Custom("POST", "/slabs/download", SlabsDownloadRequest{}, (*[]byte)(nil))
 
 	js, _ := json.Marshal(SlabsDownloadRequest{
-		Slabs:     slabs,
-		Offset:    offset,
-		Length:    length,
+		Slab:      s,
 		Contracts: contracts,
 	})
 	req, err := http.NewRequest("POST", fmt.Sprintf("%v%v", c.c.BaseURL, "/slabs/download"), bytes.NewReader(js))
@@ -422,15 +406,15 @@ func (c *Client) DownloadSlabs(dst io.Writer, slabs []slab.Slice, offset, length
 	return
 }
 
-// MigrateSlabs migrates the specified slabs.
-func (c *Client) MigrateSlabs(slabs []slab.Slab, from, to []Contract, currentHeight uint64) (err error) {
+// MigrateSlab migrates the specified slab.
+func (c *Client) MigrateSlab(s *slab.Slab, from, to []Contract, currentHeight uint64) (err error) {
 	req := SlabsMigrateRequest{
-		Slabs:         slabs,
+		Slab:          *s,
 		From:          from,
 		To:            to,
 		CurrentHeight: currentHeight,
 	}
-	err = c.c.POST("/slabs/migrate", req, nil)
+	err = c.c.POST("/slabs/migrate", req, s)
 	return
 }
 

api/server.go

@@ -88,10 +88,10 @@ type (
 
 	// A SlabMover uploads, downloads, and migrates slabs.
 	SlabMover interface {
-		UploadSlabs(ctx context.Context, r io.Reader, m, n uint8, currentHeight uint64, contracts []Contract) ([]slab.Slab, error)
-		DownloadSlabs(ctx context.Context, w io.Writer, slabs []slab.Slice, offset, length int64, contracts []Contract) error
+		UploadSlab(ctx context.Context, r io.Reader, m, n uint8, currentHeight uint64, contracts []Contract) (slab.Slab, error)
+		DownloadSlab(ctx context.Context, w io.Writer, slab slab.Slice, contracts []Contract) error
 		DeleteSlabs(ctx context.Context, slabs []slab.Slab, contracts []Contract) error
-		MigrateSlabs(ctx context.Context, slabs []slab.Slab, currentHeight uint64, from, to []Contract) error
+		MigrateSlab(ctx context.Context, s *slab.Slab, currentHeight uint64, from, to []Contract) error
 	}
 
 	// An ObjectStore stores objects.
@@ -538,21 +538,18 @@ func (s *server) hostsetsResolveHandler(jc jape.Context) {
 }
 
 func (s *server) slabsUploadHandler(jc jape.Context) {
-	jc.Custom((*[]byte)(nil), []slab.Slab{})
+	jc.Custom((*[]byte)(nil), slab.Slab{})
 
 	var sur SlabsUploadRequest
-	if err := json.NewDecoder(strings.NewReader(jc.Request.PostFormValue("meta"))).Decode(&sur); err != nil {
+	dec := json.NewDecoder(jc.Request.Body)
+	if err := dec.Decode(&sur); err != nil {
 		http.Error(jc.ResponseWriter, err.Error(), http.StatusBadRequest)
 		return
 	}
-	f, _, err := jc.Request.FormFile("data")
-	if err != nil {
-		http.Error(jc.ResponseWriter, err.Error(), http.StatusBadRequest)
-		return
-	}
-	slabs, err := s.sm.UploadSlabs(jc.Request.Context(), f, sur.MinShards, sur.TotalShards, sur.CurrentHeight, sur.Contracts)
-	if jc.Check("couldn't upload slabs", err) == nil {
-		jc.Encode(slabs)
+	data := io.LimitReader(io.MultiReader(dec.Buffered(), jc.Request.Body), int64(sur.MinShards)*rhpv2.SectorSize)
+	slab, err := s.sm.UploadSlab(jc.Request.Context(), data, sur.MinShards, sur.TotalShards, sur.CurrentHeight, sur.Contracts)
+	if jc.Check("couldn't upload slab", err) == nil {
+		jc.Encode(slab)
 	}
 }
 
@@ -563,24 +560,20 @@ func (s *server) slabsDownloadHandler(jc jape.Context) {
 	if jc.Decode(&sdr) != nil {
 		return
 	}
-	if sdr.Length == 0 {
-		for _, ss := range sdr.Slabs {
-			sdr.Length += int64(ss.Length)
-		}
-	}
-	// TODO: if we encounter an error halfway through the download, it's too
-	// late to change the response code and send an error message. Not sure how
-	// best to handle this.
-	err := s.sm.DownloadSlabs(jc.Request.Context(), jc.ResponseWriter, sdr.Slabs, sdr.Offset, sdr.Length, sdr.Contracts)
+	err := s.sm.DownloadSlab(jc.Request.Context(), jc.ResponseWriter, sdr.Slab, sdr.Contracts)
 	jc.Check("couldn't download slabs", err)
 }
 
 func (s *server) slabsMigrateHandler(jc jape.Context) {
 	var smr SlabsMigrateRequest
-	if jc.Decode(&smr) == nil {
-		err := s.sm.MigrateSlabs(jc.Request.Context(), smr.Slabs, smr.CurrentHeight, smr.From, smr.To)
-		jc.Check("couldn't migrate slabs", err)
+	if jc.Decode(&smr) != nil {
+		return
+	}
+	err := s.sm.MigrateSlab(jc.Request.Context(), &smr.Slab, smr.CurrentHeight, smr.From, smr.To)
+	if jc.Check("couldn't migrate slabs", err) != nil {
+		return
 	}
+	jc.Encode(smr.Slab)
 }
 
 func (s *server) slabsDeleteHandler(jc jape.Context) {

cmd/renterd/slab.go

@@ -39,18 +39,18 @@ func (sm slabMover) withHosts(ctx context.Context, contracts []api.Contract, fn
 	return fn(hosts)
 }
 
-func (sm slabMover) UploadSlabs(ctx context.Context, r io.Reader, m, n uint8, currentHeight uint64, contracts []api.Contract) (slabs []slab.Slab, err error) {
+func (sm slabMover) UploadSlab(ctx context.Context, r io.Reader, m, n uint8, currentHeight uint64, contracts []api.Contract) (s slab.Slab, err error) {
 	sm.pool.SetCurrentHeight(currentHeight)
 	err = sm.withHosts(ctx, contracts, func(hosts []slab.Host) error {
-		slabs, err = slab.UploadSlabs(r, m, n, hosts)
+		s, err = slab.UploadSlab(r, m, n, hosts)
 		return err
 	})
 	return
 }
 
-func (sm slabMover) DownloadSlabs(ctx context.Context, w io.Writer, slabs []slab.Slice, offset, length int64, contracts []api.Contract) error {
+func (sm slabMover) DownloadSlab(ctx context.Context, w io.Writer, s slab.Slice, contracts []api.Contract) error {
 	return sm.withHosts(ctx, contracts, func(hosts []slab.Host) error {
-		return slab.DownloadSlabs(w, slabs, offset, length, hosts)
+		return slab.DownloadSlab(w, s, hosts)
 	})
 }
 
@@ -60,7 +60,7 @@ func (sm slabMover) DeleteSlabs(ctx context.Context, slabs []slab.Slab, contract
 	})
 }
 
-func (sm slabMover) MigrateSlabs(ctx context.Context, slabs []slab.Slab, currentHeight uint64, from, to []api.Contract) (err error) {
+func (sm slabMover) MigrateSlab(ctx context.Context, s *slab.Slab, currentHeight uint64, from, to []api.Contract) (err error) {
 	sm.pool.SetCurrentHeight(currentHeight)
 	var fromHosts []slab.Host
 	for _, c := range from {
@@ -93,7 +93,7 @@ func (sm slabMover) MigrateSlabs(ctx context.Context, slabs []slab.Slab, current
 			err = ctx.Err()
 		}
 	}()
-	return slab.MigrateSlabs(slabs, fromHosts, toHosts)
+	return slab.MigrateSlab(s, fromHosts, toHosts)
 }
 
 func newSlabMover() slabMover {

cmd/renterd/web.go

@@ -133,20 +133,22 @@ func (t treeMux) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 
 func startWeb(l net.Listener, node *node, password string) error {
 	renter := api.NewServer(&syncer{node.g, node.tp}, &chainManager{node.cm}, txpool{node.tp}, node.w, node.hdb, rhpImpl{}, node.cs, newSlabMover(), node.os)
+	auth := jape.BasicAuth(password)
 	return http.Serve(l, treeMux{
 		h: createUIHandler(),
 		sub: map[string]treeMux{
-			"/api": {h: jape.AuthMiddleware(renter, password)},
+			"/api": {h: auth(renter)},
 		},
 	})
 }
 
 func startStatelessWeb(l net.Listener, password string) error {
 	renter := api.NewStatelessServer(rhpImpl{}, newSlabMover())
+	auth := jape.BasicAuth(password)
 	return http.Serve(l, treeMux{
 		h: createUIHandler(),
 		sub: map[string]treeMux{
-			"/api": {h: jape.AuthMiddleware(renter, password)},
+			"/api": {h: auth(renter)},
 		},
 	})
 }

go.mod

@@ -6,7 +6,7 @@ require (
 	github.com/hdevalence/ed25519consensus v0.0.0-20220222234857-c00d1f31bab3
 	github.com/klauspost/reedsolomon v1.9.16
 	gitlab.com/NebulousLabs/encoding v0.0.0-20200604091946-456c3dc907fe
-	go.sia.tech/jape v0.4.0
+	go.sia.tech/jape v0.5.0
 	go.sia.tech/siad v1.5.7
 	golang.org/x/crypto v0.0.0-20220525230936-793ad666bf5e
 	golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1

go.sum

@@ -147,8 +147,8 @@ gitlab.com/NebulousLabs/threadgroup v0.0.0-20200608151952-38921fbef213 h1:owERlK
 gitlab.com/NebulousLabs/threadgroup v0.0.0-20200608151952-38921fbef213/go.mod h1:vIutAvl7lmJqLVYTCBY5WDdJomP+V74At8LCeEYoH8w=
 gitlab.com/NebulousLabs/writeaheadlog v0.0.0-20200618142844-c59a90f49130/go.mod h1:SxigdS5Q1ui+OMgGAXt1E/Fg3RB6PvKXMov2O3gvIzs=
 go.etcd.io/bbolt v1.3.2/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XKL+uU=
-go.sia.tech/jape v0.4.0 h1:jB3aYycHSotYYDJs4kmR9GUfHN5NPlaTF1v+ELJnT0w=
-go.sia.tech/jape v0.4.0/go.mod h1:bu+ka8FgKq7MNH2JRTTOjfvVNku97V4ffyKr0dKoU90=
+go.sia.tech/jape v0.5.0 h1:6BLMZEePInWwQcJO1mcmrPBpF0QuvkrXmHSWGKeR0tY=
+go.sia.tech/jape v0.5.0/go.mod h1:bu+ka8FgKq7MNH2JRTTOjfvVNku97V4ffyKr0dKoU90=
 go.sia.tech/siad v1.5.7 h1:yFVNyMrCSl6vE0XjlhitFYXxgpVf15ILuCSbp7ZfExM=
 go.sia.tech/siad v1.5.7/go.mod h1:/xtHgMhNKI+cpwm5kjl9u7EG4kqMiYpmucDG710GaRY=
 go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=

internal/slabutil/mock.go

@@ -31,6 +31,8 @@ func (h *MockHost) DownloadSector(w io.Writer, root consensus.Hash256, offset, l
 	sector, ok := h.sectors[root]
 	if !ok {
 		return errors.New("unknown root")
+	} else if uint64(offset)+uint64(length) > rhpv2.SectorSize {
+		return errors.New("offset+length out of bounds")
 	}
 	_, err := w.Write(sector[offset:][:length])
 	return err