Implement lock-free DB transactions.

Lock-free write transactions allows GraphKB to have multiple instances
running in parallel to scale out.

To implement the feature I added the source breakdown in 'assets' and
'relations' tables. That way there is no concurrent access to pivot
points anymore since each data source produces and manage its own entry
in the DB. The reconciliation is done at query time once the entries are
retrieved from DB.

Author: clems4ever

go.sum

@@ -335,6 +335,7 @@ github.com/streadway/amqp v0.0.0-20190404075320-75d898a42a94/go.mod h1:AZpEONHx3
 github.com/streadway/amqp v0.0.0-20190827072141-edfb9018d271/go.mod h1:AZpEONHx3DKn8O/DFsRAY58/XVQiIPMTMB1SddzLXVw=
 github.com/streadway/handy v0.0.0-20190108123426-d5acb3125c2a/go.mod h1:qNTQ5P5JnDBl6z3cMAg/SywNDC5ABu5ApDIw6lUbRmI=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/objx v0.1.1 h1:2vfRuCMp5sSVIDSqO8oNnWJq7mPa6KVP3iPIwFBuy8A=
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0Q=

internal/database/id_generator.go

@@ -0,0 +1,66 @@
+package database
+
+import (
+	"fmt"
+
+	"github.com/clems4ever/go-graphkb/internal/knowledge"
+)
+
+// AssetTemporaryIDGenerator is a kind of cache associating IDs to assets so that pivot points are joined by those IDs
+type AssetTemporaryIDGenerator struct {
+	// map the DB ID with a temporarily generated ID.
+	DBIDToTemporaryID map[int]int
+	// map the Asset definition to the temporarily generated ID
+	DefinitionToTemporaryID map[knowledge.Asset]int
+
+	// The next ID to be used.
+	generator int
+}
+
+// NewAssetTemporaryIDGenerator create an ID Generator
+func NewAssetTemporaryIDGenerator() *AssetTemporaryIDGenerator {
+	return &AssetTemporaryIDGenerator{
+		generator:               0,
+		DBIDToTemporaryID:       make(map[int]int),
+		DefinitionToTemporaryID: make(map[knowledge.Asset]int),
+	}
+}
+
+// Push an asset into the ID generator to insert a retrieve a temporary ID
+func (atig *AssetTemporaryIDGenerator) Push(asset knowledge.Asset, DBID int) (int, error) {
+	id, ok := atig.DefinitionToTemporaryID[asset]
+	// If the asset is not in the DBID and Definition mappings, then we insert a new ID
+	if !ok {
+		_, ok2 := atig.DBIDToTemporaryID[DBID]
+		// if the DBID has not been seen yet, assign the temporary ID
+		if !ok2 {
+			id = atig.generator
+			atig.DefinitionToTemporaryID[asset] = id
+			atig.DBIDToTemporaryID[DBID] = id
+			atig.generator++
+		} else {
+			return 0, fmt.Errorf("DBID %d is already bound to another asset", DBID)
+		}
+	} else {
+		id2, ok2 := atig.DBIDToTemporaryID[DBID]
+		if ok2 && id != id2 {
+			return 0, fmt.Errorf("DBID %d is already bound to another asset", DBID)
+		}
+		atig.DBIDToTemporaryID[DBID] = id
+	}
+	return id, nil
+}
+
+// Get the temporary ID related to the given DBID
+func (atig *AssetTemporaryIDGenerator) Get(DBID int) (int, error) {
+	id, ok := atig.DBIDToTemporaryID[DBID]
+	if !ok {
+		return 0, fmt.Errorf("DB ID %d does not exist in generator", DBID)
+	}
+	return id, nil
+}
+
+// Count the number of items in the generator
+func (atig *AssetTemporaryIDGenerator) Count() int {
+	return len(atig.DBIDToTemporaryID)
+}

internal/database/id_generator_test.go

@@ -0,0 +1,84 @@
+package database
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/clems4ever/go-graphkb/internal/knowledge"
+	"github.com/clems4ever/go-graphkb/internal/schema"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func GenerateAssetHelper(idx int) knowledge.Asset {
+	return knowledge.Asset{
+		Key:  fmt.Sprintf("key%d", idx),
+		Type: schema.AssetType(fmt.Sprintf("type%d", idx)),
+	}
+}
+
+func TestGeneratedIDsAreDifferent(t *testing.T) {
+	atig := NewAssetTemporaryIDGenerator()
+
+	id1, _ := atig.Push(GenerateAssetHelper(1), 1)
+	id2, _ := atig.Push(GenerateAssetHelper(2), 2)
+
+	assert.Equal(t, 2, atig.Count())
+	assert.NotEqual(t, id1, id2)
+}
+
+func TestPushMultipleTimesAssetWithDifferentDBID(t *testing.T) {
+	atig := NewAssetTemporaryIDGenerator()
+
+	id1, _ := atig.Push(GenerateAssetHelper(2), 2)
+	id2, _ := atig.Push(GenerateAssetHelper(2), 4)
+
+	assert.Equal(t, 1, atig.Count())
+	assert.Equal(t, id1, id2)
+}
+
+func TestPushMultipleAssetsWithSameBID(t *testing.T) {
+	atig := NewAssetTemporaryIDGenerator()
+
+	atig.Push(GenerateAssetHelper(2), 2)
+	_, err := atig.Push(GenerateAssetHelper(4), 2)
+
+	assert.EqualError(t, err, "DBID 2 is already bound to another asset")
+}
+
+func TestGetExistingAssets(t *testing.T) {
+	atig := NewAssetTemporaryIDGenerator()
+
+	id1, _ := atig.Push(GenerateAssetHelper(2), 2)
+	id2, _ := atig.Push(GenerateAssetHelper(4), 4)
+
+	assert.NotEqual(t, id1, id2)
+
+	idg, err := atig.Get(2)
+	assert.NoError(t, err)
+	assert.Equal(t, id1, idg)
+
+	idg, err = atig.Get(4)
+	assert.NoError(t, err)
+	assert.Equal(t, id2, idg)
+
+	idg, err = atig.Get(5)
+	assert.EqualError(t, err, "DB ID 5 does not exist in generator")
+}
+
+func TestGetSameAsset(t *testing.T) {
+	atig := NewAssetTemporaryIDGenerator()
+
+	id1, _ := atig.Push(GenerateAssetHelper(2), 2)
+	id2, _ := atig.Push(GenerateAssetHelper(2), 4)
+
+	assert.Equal(t, id1, id2)
+
+	idg, err := atig.Get(2)
+	assert.NoError(t, err)
+	assert.Equal(t, id1, idg)
+
+	idg, err = atig.Get(4)
+	assert.NoError(t, err)
+	assert.Equal(t, id2, idg)
+}