Client: handle graph cache internally

This moves the graph caching on the library side to better handle
failure cases:
When the last tx was successful, we can reuse the last graph to avoid a
read, but when we have an error, the remote graph state is unknown, so
we must refresh it.

Author: ShimmerGlass

cmd/datasource-csv/main.go

@@ -41,13 +41,11 @@ func (cs *CSVSource) Publish() error {
 
 	r := csv.NewReader(file)
 
-	previousGraph, err := cs.graphAPI.ReadCurrentGraph()
+	tx, err := cs.graphAPI.CreateTransaction()
 	if err != nil {
-		return fmt.Errorf("Unable to read previous graph: %v", err)
+		return err
 	}
 
-	tx := cs.graphAPI.CreateTransaction(previousGraph)
-
 	header := true
 
 	for {
@@ -74,8 +72,10 @@ func (cs *CSVSource) Publish() error {
 		tx.Relate(record[1], relationType, record[4])
 	}
 
-	_, err = tx.Commit()
-	fmt.Println("CSV data has been sent successfully")
+	err = tx.Commit()
+	if err == nil {
+		fmt.Println("CSV data has been sent successfully")
+	}
 	return err
 }
 

internal/client/graph_api.go

@@ -1,6 +1,7 @@
 package client
 
 import (
+	"fmt"
 	"time"
 
 	"github.com/clems4ever/go-graphkb/internal/knowledge"
@@ -11,6 +12,8 @@ type GraphAPI struct {
 	client *GraphClient
 
 	options GraphAPIOptions
+
+	currentGraph *knowledge.Graph
 }
 
 // GraphAPIOptions options to pass to build graph API
@@ -47,7 +50,16 @@ func NewGraphAPI(options GraphAPIOptions) *GraphAPI {
 
 // CreateTransaction create a full graph transaction. This kind of transaction will diff the new graph
 // with previous version of it.
-func (gapi *GraphAPI) CreateTransaction(currentGraph *knowledge.Graph) *Transaction {
+func (gapi *GraphAPI) CreateTransaction() (*Transaction, error) {
+	if gapi.currentGraph == nil {
+		fmt.Println("transaction: fetching remote graph")
+		g, err := gapi.ReadCurrentGraph()
+		if err != nil {
+			return nil, fmt.Errorf("create transaction: %w", err)
+		}
+		gapi.currentGraph = g
+	}
+
 	var parallelization = gapi.options.Parallelization
 	if parallelization == 0 {
 		parallelization = 30
@@ -77,14 +89,28 @@ func (gapi *GraphAPI) CreateTransaction(currentGraph *knowledge.Graph) *Transact
 	transaction.newGraph = knowledge.NewGraph()
 	transaction.binder = knowledge.NewGraphBinder(transaction.newGraph)
 	transaction.client = gapi.client
-	transaction.currentGraph = currentGraph
+	transaction.currentGraph = gapi.currentGraph
 	transaction.parallelization = parallelization
 	transaction.chunkSize = chunkSize
 
 	transaction.retryCount = maxRetries
 	transaction.retryDelay = retryDelay
 	transaction.retryBackoffFactor = retryBackoff
-	return transaction
+
+	transaction.onError = func(err error) {
+		// there was an error, we don't know the remote graph state.
+		// we clear the cached copy to refresh in on the next run.
+		fmt.Println("transaction: clearing graph cache because of error:", err)
+		gapi.currentGraph = nil
+	}
+
+	transaction.onSuccess = func(g *knowledge.Graph) {
+		// tx was successful, we updated to local graph cache to
+		// speed up the next tx.
+		gapi.currentGraph = g
+	}
+
+	return transaction, nil
 }
 
 // ReadCurrentGraph read the current graph stored in graph kb

internal/client/transaction.go

@@ -36,6 +36,9 @@ type Transaction struct {
 	retryCount         int
 	retryDelay         time.Duration
 	retryBackoffFactor float64
+
+	onSuccess func(*knowledge.Graph)
+	onError   func(error)
 }
 
 // Relate create a relation between two assets
@@ -72,12 +75,14 @@ func withRetryOnTooManyRequests(fn func() error, backoffFactor float64, maxRetri
 }
 
 // Commit commit the transaction and gives ownership to the source for caching.
-func (cgt *Transaction) Commit() (*knowledge.Graph, error) {
+func (cgt *Transaction) Commit() error {
 	sg := cgt.newGraph.ExtractSchema()
 
 	fmt.Println("Start uploading the schema of the graph...")
 	if err := cgt.client.UpdateSchema(sg); err != nil {
-		return nil, fmt.Errorf("Unable to update the schema of the graph: %v", err)
+		err := fmt.Errorf("Unable to update the schema of the graph: %v", err)
+		cgt.onError(err)
+		return err
 	}
 
 	fmt.Println("Finished uploading the schema of the graph...")
@@ -141,7 +146,8 @@ func (cgt *Transaction) Commit() (*knowledge.Graph, error) {
 	for _, f := range futures {
 		err := <-f
 		if err != nil {
-			return nil, err
+			cgt.onError(err)
+			return err
 		}
 	}
 
@@ -181,13 +187,14 @@ func (cgt *Transaction) Commit() (*knowledge.Graph, error) {
 	for _, f := range futures {
 		err := <-f
 		if err != nil {
-			return nil, err
+			cgt.onError(err)
+			return err
 		}
 	}
 
 	fmt.Println("Finished uploading the graph...")
 
-	g := cgt.newGraph
+	cgt.onSuccess(cgt.newGraph)
 	cgt.newGraph = knowledge.NewGraph()
-	return g, nil // give ownership of the transaction graph so that it can be cached if needed
+	return nil
 }