diff --git a/jobs/check_alpha_payments.go b/jobs/check_alpha_payments.go index 3bd01770..29b427ef 100644 --- a/jobs/check_alpha_payments.go +++ b/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). diff --git a/models/broker_broker_transactions_test.go b/models/broker_broker_transactions_test.go index b0df288f..c1308a41 100644 --- a/models/broker_broker_transactions_test.go +++ b/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 ( diff --git a/models/upload_sessions.go b/models/upload_sessions.go index eeb6e6de..c871201b 100644 --- a/models/upload_sessions.go +++ b/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 diff --git a/models/upload_sessions_test.go b/models/upload_sessions_test.go index a60c62e0..0c89a092 100644 --- a/models/upload_sessions_test.go +++ b/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() { diff --git a/services/iota_wrappers.go b/services/iota_wrappers.go index a0f0fa5c..badb98e2 100644 --- a/services/iota_wrappers.go +++ b/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) } diff --git a/utils/kv_store.go b/utils/kv_store.go index 09c70a26..402e999b 100644 --- a/utils/kv_store.go +++ b/utils/kv_store.go @@ -2,7 +2,6 @@ package oyster_utils import ( "errors" - "github.com/orcaman/concurrent-map" "io/ioutil" "os" "strconv" @@ -10,6 +9,7 @@ import ( "time" "github.com/dgraph-io/badger" + "github.com/orcaman/concurrent-map" ) // const badgerDir = "/tmp/badger" // TODO: CHANGE THIS.