Prioritize meta

jobs/check_alpha_payments.go

@@ -1,11 +1,12 @@
 package jobs
 
 import (
+	"math/big"
+
 	"github.com/oysterprotocol/brokernode/models"
 	"github.com/oysterprotocol/brokernode/services"
 	"github.com/oysterprotocol/brokernode/utils"
 	"gopkg.in/segmentio/analytics-go.v3"
-	"math/big"
 )
 
 /* CheckAlphaPayments handles the operations around checking the status of payments to alpha and
@@ -46,6 +47,19 @@ func CheckPaymentToAlpha() {
 			}
 
 			models.SetUploadSessionToPaid(brokerTx)
+
+			// Attach metachunk.
+			go func() {
+				metaChunk, err := models.GetMetaChunk(brokerTx.GenesisHash)
+				oyster_utils.LogIfError(err, nil)
+				if err != nil {
+					return
+				}
+
+				err = services.PowAndBroadcast([]oyster_utils.ChunkData{metaChunk})
+				oyster_utils.LogIfError(err, nil)
+			}()
+
 			oyster_utils.LogToSegment("check_alpha_payments: CheckPaymentToAlpha - alpha_confirmed",
 				analytics.NewProperties().
 					Set("beta_address", brokerTx.ETHAddrBeta).

models/broker_broker_transactions_test.go

@@ -1,12 +1,13 @@
 package models_test
 
 import (
+	"time"
+
 	"github.com/gobuffalo/pop/nulls"
 	"github.com/oysterprotocol/brokernode/jobs"
 	"github.com/oysterprotocol/brokernode/models"
 	"github.com/oysterprotocol/brokernode/utils"
 	"github.com/shopspring/decimal"
-	"time"
 )
 
 var (

models/upload_sessions.go

@@ -7,7 +7,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"golang.org/x/crypto/sha3"
 	"math"
 	"math/big"
 	"math/rand"
@@ -22,6 +21,7 @@ import (
 	"github.com/gobuffalo/validate/validators"
 	"github.com/oysterprotocol/brokernode/utils"
 	"github.com/shopspring/decimal"
+	"golang.org/x/crypto/sha3"
 )
 
 /*ChunkReq is the form in which webinterface will send the data for each chunk*/
@@ -753,6 +753,24 @@ func checkIfAllMessagesAreReadyInSQL(treasureIndexes []int, u *UploadSession) bo
 	return allMessagesFound
 }
 
+// GetMetaChunk will fetch the metachunk associated with the genHash passed in.
+func GetMetaChunk(genHash string) (oyster_utils.ChunkData, error) {
+	metaIdx := int64(0) // rev1
+	stopIdx := metaIdx + 1
+
+	keys := oyster_utils.GenerateBulkKeys(genHash, metaIdx, stopIdx)
+	chunks, err := GetMultiChunkData(oyster_utils.InProgressDir, genHash, keys)
+	oyster_utils.LogIfError(err, nil)
+
+	if len(chunks) > 0 {
+		metaChunk := chunks[0] // Fetched just 1 chunk.
+		return metaChunk, err
+	}
+
+	return oyster_utils.ChunkData{}, errors.New("metachunk not found")
+
+}
+
 /*GetUnassignedChunksBySession returns the chunk data for chunks that need attaching for a particular session*/
 func (u *UploadSession) GetUnassignedChunksBySession(offset int) (chunkData []oyster_utils.ChunkData, err error) {
 	var stopChunkIdx int64

models/upload_sessions_test.go

@@ -3,13 +3,14 @@ package models_test
 import (
 	"encoding/hex"
 	"fmt"
+	"math/big"
+	"strconv"
+	"time"
+
 	"github.com/gobuffalo/pop/nulls"
 	"github.com/oysterprotocol/brokernode/models"
 	"github.com/oysterprotocol/brokernode/utils"
 	"github.com/shopspring/decimal"
-	"math/big"
-	"strconv"
-	"time"
 )
 
 func (suite *ModelSuite) Test_BigFileSize() {

services/iota_wrappers.go

@@ -186,51 +186,59 @@ func init() {
 	PoWFrequency.Frequency = 2
 }
 
-func PowWorker(jobQueue <-chan PowJob, channelID string, err error) {
-	for powJobRequest := range jobQueue {
-		// this is where we would call methods to deal with each job request
-		fmt.Println("PowWorker: Starting")
+// PowAndBroadcast does PoW then attaches to the tangle
+func PowAndBroadcast(chunks []oyster_utils.ChunkData) (err error) {
+	transfersArray := make([]giota.Transfer, len(chunks))
 
-		startTime := time.Now()
+	for i, chunk := range chunks {
+		address, err := giota.ToAddress(chunk.Address)
+		if err != nil {
+			oyster_utils.LogIfError(err, nil)
+			panic(err)
+		}
+		transfersArray[i].Address = address
+		transfersArray[i].Value = int64(0)
+		transfersArray[i].Message, err = giota.ToTrytes(chunk.Message)
+		if err != nil {
+			oyster_utils.LogIfError(err, nil)
+			panic(err)
+		}
+		transfersArray[i].Tag = OysterTag
+	}
 
-		transfersArray := make([]giota.Transfer, len(powJobRequest.Chunks))
+	bdl, err := giota.PrepareTransfers(api, seed, transfersArray, nil, "", 1)
+	if err != nil {
+		return err
+	}
 
-		for i, chunk := range powJobRequest.Chunks {
-			address, err := giota.ToAddress(chunk.Address)
-			if err != nil {
-				oyster_utils.LogIfError(err, nil)
-				panic(err)
-			}
-			transfersArray[i].Address = address
-			transfersArray[i].Value = int64(0)
-			transfersArray[i].Message, err = giota.ToTrytes(chunk.Message)
-			if err != nil {
-				oyster_utils.LogIfError(err, nil)
-				panic(err)
-			}
-			transfersArray[i].Tag = OysterTag
-		}
+	transactions := []giota.Transaction(bdl)
+	transactionsToApprove, err := getTransactionsToApprove()
+	if err != nil {
+		return err
+	}
 
-		bdl, err := giota.PrepareTransfers(api, seed, transfersArray, nil, "", 1)
+	err = doPowAndBroadcast(
+		transactionsToApprove.BranchTransaction,
+		transactionsToApprove.TrunkTransaction,
+		minDepth,
+		transactions,
+		minWeightMag,
+		bestPow,
+	)
 
-		oyster_utils.LogIfError(err, nil)
+	return err
+}
 
-		transactions := []giota.Transaction(bdl)
+func PowWorker(jobQueue <-chan PowJob, channelID string, err error) {
+	for powJobRequest := range jobQueue {
+		// this is where we would call methods to deal with each job request
+		fmt.Println("PowWorker: Starting")
+		startTime := time.Now()
 
-		transactionsToApprove, err := getTransactionsToApprove()
+		err = PowAndBroadcast(powJobRequest.Chunks)
 		oyster_utils.LogIfError(err, nil)
 
 		if err == nil {
-
-			err = doPowAndBroadcast(
-				transactionsToApprove.BranchTransaction,
-				transactionsToApprove.TrunkTransaction,
-				minDepth,
-				transactions,
-				minWeightMag,
-				bestPow,
-				powJobRequest.BroadcastNodes)
-
 			channelToChange := Channel[channelID]
 
 			channelInDB := models.ChunkChannel{}
@@ -316,7 +324,7 @@ func findTransactions(addresses []giota.Address) (map[giota.Address][]giota.Tran
 }
 
 func doPowAndBroadcast(branch giota.Trytes, trunk giota.Trytes, depth int64,
-	trytes []giota.Transaction, mwm int64, bestPow giota.PowFunc, broadcastNodes []string) error {
+	trytes []giota.Transaction, mwm int64, bestPow giota.PowFunc) error {
 
 	defer oyster_utils.TimeTrack(time.Now(), "iota_wrappers: doPow_using_"+powName, analytics.NewProperties().
 		//Set("addresses", oyster_utils.MapTransactionsToAddrs(trytes)))
@@ -380,7 +388,6 @@ func doPowAndBroadcast(branch giota.Trytes, trunk giota.Trytes, depth int64,
 }
 
 func sendChunksToLambda(chunks *[]oyster_utils.ChunkData) {
-
 	go batchPowOnLambda(chunks)
 }
 

utils/kv_store.go

@@ -2,14 +2,14 @@ package oyster_utils
 
 import (
 	"errors"
-	"github.com/orcaman/concurrent-map"
 	"io/ioutil"
 	"os"
 	"strconv"
 	"strings"
 	"time"
 
 	"github.com/dgraph-io/badger"
+	"github.com/orcaman/concurrent-map"
 )
 
 // const badgerDir = "/tmp/badger" // TODO: CHANGE THIS.