core: ObjectStorage.Begin and TxObjectStorage

Author: mcuadros

blame_test.go

@@ -31,8 +31,8 @@ func (s *BlameCommon) SetUpSuite(c *C) {
 
 		stream := packfile.NewStream(bytes.NewReader(data))
 
-		d := packfile.NewDecoder(stream)
-		err = d.Decode(r.s.ObjectStorage())
+		d := packfile.NewDecoder(stream, r.s.ObjectStorage())
+		err = d.Decode()
 		c.Assert(err, IsNil)
 
 		c.Assert(f.Close(), IsNil)

common_test.go

@@ -123,8 +123,8 @@ func unpackFixtures(c *C, fixtures ...[]packedFixture) map[string]*Repository {
 			memStream := bytes.NewReader(data)
 			r := packfile.NewStream(memStream)
 
-			d := packfile.NewDecoder(r)
-			err = d.Decode(repos[fixture.url].s.ObjectStorage())
+			d := packfile.NewDecoder(r, repos[fixture.url].s.ObjectStorage())
+			err = d.Decode()
 			c.Assert(err, IsNil, comment)
 
 			c.Assert(f.Close(), IsNil, comment)

core/storage.go

@@ -7,19 +7,28 @@ var ErrStop = errors.New("stop iter")
 
 // ObjectStorage generic storage of objects
 type ObjectStorage interface {
+	// NewObject returns a new Object, the real type of the object can be a
+	// custom implementation or the defaul one, MemoryObject
 	NewObject() Object
+	// Set save an object into the storage, the object shuld be create with
+	// the NewObject, method, and file if the type is not supported.
 	Set(Object) (Hash, error)
-	// Get an object by hash with the given ObjectType.
-	//
-	// Implementors should return (nil, core.ErrObjectNotFound) if an object
-	// doesn't exist with both the given hash and object type.
+	// Get an object by hash with the given ObjectType. Implementors should
+	// return (nil, ErrObjectNotFound) if an object doesn't exist with both the
+	// given hash and object type.
 	//
 	// Valid ObjectType values are CommitObject, BlobObject, TagObject, TreeObject
 	// and AnyObject.
 	//
 	// If AnyObject is given, the object must be looked up regardless of its type.
 	Get(ObjectType, Hash) (Object, error)
+	// Iter returns a custom ObjectIter over all the object on the storage.
+	//
+	// Valid ObjectType values are CommitObject, BlobObject, TagObject, TreeObject
+	// and AnyObject.
 	Iter(ObjectType) (ObjectIter, error)
+	// Begin starts a transaction.
+	Begin() TxObjectStorage
 }
 
 // ObjectIter is a generic closable interface for iterating over objects.
@@ -29,6 +38,14 @@ type ObjectIter interface {
 	Close()
 }
 
+// TxObjectStorage is an in-progress storage transaction.
+// A transaction must end with a call to Commit or Rollback.
+type TxObjectStorage interface {
+	Set(Object) (Hash, error)
+	Commit() error
+	Rollback() error
+}
+
 // ReferenceStorage generic storage of references
 type ReferenceStorage interface {
 	Set(*Reference) error

examples/storage/aerospike/storage.go

@@ -131,6 +131,26 @@ func (s *ObjectStorage) buildKey(h core.Hash, t core.ObjectType) (*driver.Key, e
 	return driver.NewKey(s.ns, t.String(), fmt.Sprintf("%s|%s", s.url, h.String()))
 }
 
+func (s *ObjectStorage) Begin() core.TxObjectStorage {
+	return &TxObjectStorage{Storage: s}
+}
+
+type TxObjectStorage struct {
+	Storage *ObjectStorage
+}
+
+func (tx *TxObjectStorage) Set(obj core.Object) (core.Hash, error) {
+	return tx.Storage.Set(obj)
+}
+
+func (tx *TxObjectStorage) Commit() error {
+	return nil
+}
+
+func (tx *TxObjectStorage) Rollback() error {
+	return nil
+}
+
 type ObjectIter struct {
 	t  core.ObjectType
 	ch chan *driver.Record

formats/packfile/decoder.go

@@ -42,25 +42,33 @@ type Decoder struct {
 }
 
 // NewDecoder returns a new Decoder that reads from r.
-func NewDecoder(r ReadRecaller) *Decoder {
+func NewDecoder(r ReadRecaller, s core.ObjectStorage) *Decoder {
 	return &Decoder{
 		p: NewParser(r),
+		s: s,
 	}
 }
 
 // Decode reads a packfile and stores it in the value pointed to by s.
-func (d *Decoder) Decode(s core.ObjectStorage) error {
-	d.s = s
-
+func (d *Decoder) Decode() error {
 	count, err := d.p.ReadHeader()
 	if err != nil {
 		return err
 	}
 
-	return d.readObjects(count)
+	tx := d.s.Begin()
+	if err := d.readObjects(tx, count); err != nil {
+		if err := tx.Rollback(); err != nil {
+			return nil
+		}
+
+		return err
+	}
+
+	return tx.Commit()
 }
 
-func (d *Decoder) readObjects(count uint32) error {
+func (d *Decoder) readObjects(tx core.TxObjectStorage, count uint32) error {
 	// This code has 50-80 µs of overhead per object not counting zlib inflation.
 	// Together with zlib inflation, it's 400-410 µs for small objects.
 	// That's 1 sec for ~2450 objects, ~4.20 MB, or ~250 ms per MB,
@@ -85,7 +93,7 @@ func (d *Decoder) readObjects(count uint32) error {
 			return err
 		}
 
-		_, err = d.s.Set(obj)
+		_, err = tx.Set(obj)
 		if err == io.EOF {
 			break
 		}

formats/packfile/decoder_test.go

@@ -28,10 +28,10 @@ func (s *ReaderSuite) TestReadPackfile(c *C) {
 	data, _ := base64.StdEncoding.DecodeString(packFileWithEmptyObjects)
 	f := bytes.NewReader(data)
 	r := NewStream(f)
-	d := NewDecoder(r)
-
 	sto := memory.NewStorage()
-	err := d.Decode(sto.ObjectStorage())
+	d := NewDecoder(r, sto.ObjectStorage())
+
+	err := d.Decode()
 	c.Assert(err, IsNil)
 
 	AssertObjects(c, sto, []string{
@@ -61,10 +61,10 @@ func (s *ReaderSuite) testReadPackfileGitFixture(c *C, file string, format Forma
 	f, err := os.Open(file)
 	c.Assert(err, IsNil)
 	r := NewSeekable(f)
-	d := NewDecoder(r)
-
 	sto := memory.NewStorage()
-	err = d.Decode(sto.ObjectStorage())
+	d := NewDecoder(r, sto.ObjectStorage())
+
+	err = d.Decode()
 	c.Assert(err, IsNil)
 
 	AssertObjects(c, sto, []string{
@@ -170,10 +170,10 @@ func readFromFile(c *C, file string, format Format) *memory.ObjectStorage {
 	f, err := os.Open(file)
 	c.Assert(err, IsNil)
 	r := NewSeekable(f)
-	d := NewDecoder(r)
-
 	sto := memory.NewStorage()
-	err = d.Decode(sto.ObjectStorage())
+	d := NewDecoder(r, sto.ObjectStorage())
+
+	err = d.Decode()
 	c.Assert(err, IsNil)
 
 	return sto.ObjectStorage().(*memory.ObjectStorage)

references_test.go

@@ -27,8 +27,8 @@ func (s *ReferencesSuite) SetUpSuite(c *C) {
 		defer f.Close()
 		c.Assert(err, IsNil)
 		r := packfile.NewSeekable(f)
-		d := packfile.NewDecoder(r)
-		err = d.Decode(s.repos[fix.url].s.ObjectStorage())
+		d := packfile.NewDecoder(r, s.repos[fix.url].s.ObjectStorage())
+		err = d.Decode()
 		c.Assert(err, IsNil)
 	}
 }

remote.go

@@ -163,8 +163,8 @@ func (r *Remote) buildRequest(
 func (r *Remote) updateObjectStorage(reader io.Reader) error {
 	stream := packfile.NewStream(reader)
 
-	d := packfile.NewDecoder(stream)
-	return d.Decode(r.s.ObjectStorage())
+	d := packfile.NewDecoder(stream, r.s.ObjectStorage())
+	return d.Decode()
 }
 
 func (r *Remote) updateLocalReferenceStorage(specs []config.RefSpec, refs []*core.Reference) error {

storage/filesystem/object.go

@@ -217,3 +217,20 @@ func buildIndexFromIdxfile(fs fs.FS, path string) (index.Index, error) {
 
 	return index.NewFromIdx(f)
 }
+func (o *ObjectStorage) Begin() core.TxObjectStorage {
+	return &TxObjectStorage{}
+}
+
+type TxObjectStorage struct{}
+
+func (tx *TxObjectStorage) Set(obj core.Object) (core.Hash, error) {
+	return core.ZeroHash, fmt.Errorf("not implemented yet")
+}
+
+func (tx *TxObjectStorage) Commit() error {
+	return fmt.Errorf("not implemented yet")
+}
+
+func (tx *TxObjectStorage) Rollback() error {
+	return fmt.Errorf("not implemented yet")
+}

storage/filesystem/object_test.go

@@ -124,9 +124,9 @@ func memStorageFromGitDir(fs fs.FS, path string) (core.ObjectStorage, error) {
 
 	sto := memory.NewStorage()
 	r := packfile.NewStream(f)
-	d := packfile.NewDecoder(r)
+	d := packfile.NewDecoder(r, sto.ObjectStorage())
 
-	err = d.Decode(sto.ObjectStorage())
+	err = d.Decode()
 	if err != nil {
 		return nil, err
 	}
@@ -267,8 +267,8 @@ func memStorageFromDirPath(fs fs.FS, path string) (core.ObjectStorage, error) {
 	}
 
 	r := packfile.NewStream(f)
-	d := packfile.NewDecoder(r)
-	err = d.Decode(sto.ObjectStorage())
+	d := packfile.NewDecoder(r, sto.ObjectStorage())
+	err = d.Decode()
 	if err != nil {
 		return nil, err
 	}