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

satellite/metainfo: avoid error on inserting existing attribution

Browse Source 
Currently we can have an error about duplicated entry while inserting into value_attributions table. This change is changing simple insert into insert that is doing nothing on conflict.

Change-Id: I3efd8dc0b63115e8e2ed8f4196ccf969ee942295
This commit is contained in:
Michał Niewrzał 2021-04-01 16:50:53 +02:00 committed by Michal Niewrzal
parent 5c038c4325
commit 71d0adf90b
2 changed files with 56 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
satellite/metainfo/attribution_test.go
View File

@ -329,3 +329,48 @@ func TestAttributionReport(t *testing.T) {
}
})
}
func TestBucketAttributionConcurrentUpload(t *testing.T) {
testplanet.Run(t, testplanet.Config{
SatelliteCount: 1,
StorageNodeCount: 0,
UplinkCount: 1,
}, func(t *testing.T, ctx *testcontext.Context, planet *testplanet.Planet) {
satellite := planet.Satellites[0]
err := planet.Uplinks[0].CreateBucket(ctx, satellite, "attr-bucket")
require.NoError(t, err)
config := uplink.Config{
UserAgent: "Minio",
}
project, err := config.OpenProject(ctx, planet.Uplinks[0].Access[satellite.ID()])
require.NoError(t, err)
for i := 0; i < 3; i++ {
i := i
ctx.Go(func() error {
upload, err := project.UploadObject(ctx, "attr-bucket", "key"+strconv.Itoa(i), nil)
require.NoError(t, err)
_, err = upload.Write([]byte("content"))
require.NoError(t, err)
err = upload.Commit()
require.NoError(t, err)
return nil
})
}
ctx.Wait()
expectedPartnerID, err := satellite.Metainfo.Endpoint2.ResolvePartnerID(ctx, &pb.RequestHeader{
UserAgent: []byte("Minio"),
})
require.NoError(t, err)
attributionInfo, err := planet.Satellites[0].DB.Attribution().Get(ctx, planet.Uplinks[0].Projects[0].ID, []byte("attr-bucket"))
require.NoError(t, err)
require.Equal(t, expectedPartnerID, attributionInfo.PartnerID)
})
}

17
satellite/satellitedb/attribution.go
View File

@ -135,16 +135,21 @@ func (keys *attributionDB) Get(ctx context.Context, projectID uuid.UUID, bucketN
func (keys *attributionDB) Insert(ctx context.Context, info *attribution.Info) (_ *attribution.Info, err error) {
defer mon.Task()(&ctx)(&err)
dbxInfo, err := keys.db.Create_ValueAttribution(ctx,
dbx.ValueAttribution_ProjectId(info.ProjectID[:]),
dbx.ValueAttribution_BucketName(info.BucketName),
dbx.ValueAttribution_PartnerId(info.PartnerID[:]),
)
err = keys.db.QueryRowContext(ctx, `
INSERT INTO value_attributions (project_id, bucket_name, partner_id, last_updated)
VALUES ($1, $2, $3, now())
ON CONFLICT (project_id, bucket_name) DO NOTHING
RETURNING last_updated
`, info.ProjectID[:], info.BucketName, info.PartnerID[:]).Scan(&info.CreatedAt)
// TODO when sql.ErrNoRows is returned then CreatedAt is not set
if errors.Is(err, sql.ErrNoRows) {
return info, nil
}
if err != nil {
return nil, Error.Wrap(err)
}
return attributionFromDBX(dbxInfo)
return info, nil
}
// QueryAttribution queries partner bucket attribution data.