JakeHillion/storj
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

satellite/satellitedb: use transaction helpers in containment

Browse Source 
Transactions in our code that might need to work against CockroachDB
need to be retried in the event of a retryable error. The transaction
helper functions in dbutil do that automatically. I am changing this
code to use those helpers instead.

Change-Id: I660540885a0784fae844cf99376d1537e208fa69
This commit is contained in:
paul cannon 2019-12-19 03:52:17 -06:00 committed by paul cannon
parent 2549c601e9
commit f3aee1b758
1 changed files with 43 additions and 57 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

100
satellite/satellitedb/containment.go
View File

@ -8,8 +8,6 @@ import (
"context"
"database/sql"
"github.com/zeebo/errs"
"storj.io/common/pb"
"storj.io/common/storj"
"storj.io/storj/satellite/audit"
@ -41,76 +39,64 @@ func (containment *containment) Get(ctx context.Context, id pb.NodeID) (_ *audit
// IncrementPending creates a new pending audit entry, or increases its reverify count if it already exists
func (containment *containment) IncrementPending(ctx context.Context, pendingAudit *audit.PendingAudit) (err error) {
defer mon.Task()(&ctx)(&err)
tx, err := containment.db.Open(ctx)
if err != nil {
return audit.ContainError.Wrap(err)
}
existingAudit, err := tx.Get_PendingAudits_By_NodeId(ctx, dbx.PendingAudits_NodeId(pendingAudit.NodeID.Bytes()))
switch err {
case sql.ErrNoRows:
statement := containment.db.Rebind(
`INSERT INTO pending_audits (node_id, piece_id, stripe_index, share_size, expected_share_hash, reverify_count, path)
err = containment.db.WithTx(ctx, func(ctx context.Context, tx *dbx.Tx) error {
existingAudit, err := tx.Get_PendingAudits_By_NodeId(ctx, dbx.PendingAudits_NodeId(pendingAudit.NodeID.Bytes()))
switch err {
case sql.ErrNoRows:
statement := containment.db.Rebind(
`INSERT INTO pending_audits (node_id, piece_id, stripe_index, share_size, expected_share_hash, reverify_count, path)
VALUES (?, ?, ?, ?, ?, ?, ?)`,
)
_, err = tx.Tx.ExecContext(ctx, statement, pendingAudit.NodeID.Bytes(), pendingAudit.PieceID.Bytes(), pendingAudit.StripeIndex,
pendingAudit.ShareSize, pendingAudit.ExpectedShareHash, pendingAudit.ReverifyCount, []byte(pendingAudit.Path))
if err != nil {
return audit.ContainError.Wrap(errs.Combine(err, tx.Rollback()))
}
case nil:
if !bytes.Equal(existingAudit.ExpectedShareHash, pendingAudit.ExpectedShareHash) {
return audit.ContainError.Wrap(errs.Combine(audit.ErrAlreadyExists.New("%v", pendingAudit.NodeID), tx.Rollback()))
}
statement := containment.db.Rebind(
`UPDATE pending_audits SET reverify_count = pending_audits.reverify_count + 1
)
_, err = tx.Tx.ExecContext(ctx, statement, pendingAudit.NodeID.Bytes(), pendingAudit.PieceID.Bytes(), pendingAudit.StripeIndex,
pendingAudit.ShareSize, pendingAudit.ExpectedShareHash, pendingAudit.ReverifyCount, []byte(pendingAudit.Path))
if err != nil {
return err
}
case nil:
if !bytes.Equal(existingAudit.ExpectedShareHash, pendingAudit.ExpectedShareHash) {
return audit.ErrAlreadyExists.New("%v", pendingAudit.NodeID)
}
statement := containment.db.Rebind(
`UPDATE pending_audits SET reverify_count = pending_audits.reverify_count + 1
WHERE pending_audits.node_id=?`,
)
_, err = tx.Tx.ExecContext(ctx, statement, pendingAudit.NodeID.Bytes())
if err != nil {
return audit.ContainError.Wrap(errs.Combine(err, tx.Rollback()))
)
_, err = tx.Tx.ExecContext(ctx, statement, pendingAudit.NodeID.Bytes())
if err != nil {
return err
}
default:
return err
}
default:
return audit.ContainError.Wrap(errs.Combine(err, tx.Rollback()))
}
updateContained := dbx.Node_Update_Fields{
Contained: dbx.Node_Contained(true),
}
updateContained := dbx.Node_Update_Fields{
Contained: dbx.Node_Contained(true),
}
err = tx.UpdateNoReturn_Node_By_Id(ctx, dbx.Node_Id(pendingAudit.NodeID.Bytes()), updateContained)
if err != nil {
return audit.ContainError.Wrap(errs.Combine(err, tx.Rollback()))
}
return audit.ContainError.Wrap(tx.Commit())
return tx.UpdateNoReturn_Node_By_Id(ctx, dbx.Node_Id(pendingAudit.NodeID.Bytes()), updateContained)
})
return audit.ContainError.Wrap(err)
}
// Delete deletes the pending audit
func (containment *containment) Delete(ctx context.Context, id pb.NodeID) (_ bool, err error) {
func (containment *containment) Delete(ctx context.Context, id pb.NodeID) (isDeleted bool, err error) {
defer mon.Task()(&ctx)(&err)
if id.IsZero() {
return false, audit.ContainError.New("node ID empty")
}
tx, err := containment.db.Open(ctx)
if err != nil {
return false, audit.ContainError.Wrap(err)
}
err = containment.db.WithTx(ctx, func(ctx context.Context, tx *dbx.Tx) (err error) {
isDeleted, err = tx.Delete_PendingAudits_By_NodeId(ctx, dbx.PendingAudits_NodeId(id.Bytes()))
if err != nil {
return err
}
isDeleted, err := tx.Delete_PendingAudits_By_NodeId(ctx, dbx.PendingAudits_NodeId(id.Bytes()))
if err != nil {
return isDeleted, audit.ContainError.Wrap(errs.Combine(audit.ErrContainDelete.Wrap(err), tx.Rollback()))
}
updateContained := dbx.Node_Update_Fields{
Contained: dbx.Node_Contained(false),
}
updateContained := dbx.Node_Update_Fields{
Contained: dbx.Node_Contained(false),
}
err = tx.UpdateNoReturn_Node_By_Id(ctx, dbx.Node_Id(id.Bytes()), updateContained)
if err != nil {
return isDeleted, audit.ContainError.Wrap(errs.Combine(err, tx.Rollback()))
}
return isDeleted, audit.ContainError.Wrap(tx.Commit())
return tx.UpdateNoReturn_Node_By_Id(ctx, dbx.Node_Id(id.Bytes()), updateContained)
})
return isDeleted, audit.ContainError.Wrap(err)
}
func convertDBPending(ctx context.Context, info *dbx.PendingAudits) (_ *audit.PendingAudit, err error) {