Skip to content

Commit cc68a9c

Browse files
ChengxuanChengxuan
committed
Merge branch 'main' of github.com:hyperledger/firefly-transaction-manager into pass-in-action-occurred-time
Signed-off-by: Chengxuan Xing <[email protected]>
2 parents cd9755d + af4a223 commit cc68a9c

15 files changed

+225
-33
lines changed

‎Makefile

+1-1
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,7 @@ coverage.html:
1717
$(VGO) tool cover -html=coverage.txt
1818
coverage: test coverage.html
1919
lint:
20-
$(VGO) install github.com/golangci/golangci-lint/cmd/golangci-lint@v1.53.2
20+
$(VGO) install github.com/golangci/golangci-lint/cmd/golangci-lint@v1.55.2
2121
GOGC=20 $(LINT) run -v --timeout 5m
2222

2323
${MOCKERY}:

‎db/migrations/postgres/000009_add_transactions_status_index.down.sql

+1
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
DROP INDEX IF EXISTS transactions_status;

‎db/migrations/postgres/000009_add_transactions_status_index.up.sql

+1
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
CREATE INDEX transactions_status ON transactions(status);

‎go.mod

+1-1
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ require (
1212
github.com/gorilla/mux v1.8.1
1313
github.com/gorilla/websocket v1.5.1
1414
github.com/hashicorp/golang-lru/v2 v2.0.7
15-
github.com/hyperledger/firefly-common v1.4.2
15+
github.com/hyperledger/firefly-common v1.4.6
1616
github.com/lib/pq v1.10.9
1717
github.com/oklog/ulid/v2 v2.1.0
1818
github.com/prometheus/client_golang v1.18.0

‎go.sum

+2-2
Original file line numberDiff line numberDiff line change
@@ -90,8 +90,8 @@ github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpO
9090
github.com/huandu/xstrings v1.3.3/go.mod h1:y5/lhBue+AyNmUVz9RLU9xbLR0o4KIIExikq4ovT0aE=
9191
github.com/huandu/xstrings v1.4.0 h1:D17IlohoQq4UcpqD7fDk80P7l+lwAmlFaBHgOipl2FU=
9292
github.com/huandu/xstrings v1.4.0/go.mod h1:y5/lhBue+AyNmUVz9RLU9xbLR0o4KIIExikq4ovT0aE=
93-
github.com/hyperledger/firefly-common v1.4.2 h1:sBbiTFWDu1qCnXFA6JobasJl4AXphCAUZU/R4nyWPdE=
94-
github.com/hyperledger/firefly-common v1.4.2/go.mod h1:jkErZdQmC9fsAJZQO427tURdwB9iiW+NMUZSqS3eBIE=
93+
github.com/hyperledger/firefly-common v1.4.6 h1:qqXoSaRml3WjUnWcWxrrXs5AIOWa+UcMXLCF8yEa4Pk=
94+
github.com/hyperledger/firefly-common v1.4.6/go.mod h1:jkErZdQmC9fsAJZQO427tURdwB9iiW+NMUZSqS3eBIE=
9595
github.com/imdario/mergo v0.3.11/go.mod h1:jmQim1M+e3UYxmgPu/WyfjB3N3VflVyUjjjwH0dnCYA=
9696
github.com/imdario/mergo v0.3.16 h1:wwQJbIsHYGMUyLSPrEq1CT16AhnhNJQ51+4fdHUnCl4=
9797
github.com/imdario/mergo v0.3.16/go.mod h1:WBLT9ZmE3lPoWsEzCh9LPo3TiwVN+ZKEjmz+hD27ysY=

‎internal/persistence/postgres/eventstreams_test.go

+5-2
Original file line numberDiff line numberDiff line change
@@ -108,8 +108,11 @@ func TestEventStreamAfterPaginatePSQL(t *testing.T) {
108108
var eventStreams []*apitypes.EventStream
109109
for i := 0; i < 20; i++ {
110110
es := &apitypes.EventStream{
111-
ID: fftypes.NewUUID(),
112-
Name: strPtr(fmt.Sprintf("es_%.3d", i)),
111+
ID: fftypes.NewUUID(),
112+
Name: strPtr(fmt.Sprintf("es_%.3d", i)),
113+
BatchTimeout: ffDurationPtr(22222 * time.Second),
114+
RetryTimeout: ffDurationPtr(33333 * time.Second),
115+
BlockedRetryDelay: ffDurationPtr(44444 * time.Second),
113116
}
114117
err := p.WriteStream(ctx, es)
115118
assert.NoError(t, err)

‎internal/persistence/postgres/transaction_writer.go

+25-5
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
// Copyright © 2023 Kaleido, Inc.
1+
// Copyright © 2024 Kaleido, Inc.
22
//
33
// SPDX-License-Identifier: Apache-2.0
44
//
@@ -40,6 +40,7 @@ type transactionOperation struct {
4040
sentConflict bool
4141
done chan error
4242

43+
opID string
4344
isShutdown bool
4445
txInsert *apitypes.ManagedTX
4546
noncePreAssigned bool
@@ -77,6 +78,7 @@ type transactionWriter struct {
7778

7879
type transactionWriterBatch struct {
7980
id string
81+
opened time.Time
8082
ops []*transactionOperation
8183
timeoutContext context.Context
8284
timeoutCancel func()
@@ -122,6 +124,7 @@ func newTransactionWriter(bgCtx context.Context, p *sqlPersistence, conf config.
122124

123125
func newTransactionOperation(txID string) *transactionOperation {
124126
return &transactionOperation{
127+
opID: fftypes.ShortID(),
125128
txID: txID,
126129
done: make(chan error, 1), // 1 slot to ensure we don't block the writer
127130
}
@@ -130,6 +133,7 @@ func newTransactionOperation(txID string) *transactionOperation {
130133
func (op *transactionOperation) flush(ctx context.Context) error {
131134
select {
132135
case err := <-op.done:
136+
log.L(ctx).Debugf("Flushed write operation %s (err=%v)", op.opID, err)
133137
return err
134138
case <-ctx.Done():
135139
return i18n.NewError(ctx, i18n.MsgContextCanceled)
@@ -165,6 +169,7 @@ func (tw *transactionWriter) queue(ctx context.Context, op *transactionOperation
165169
h := fnv.New32a() // simple non-cryptographic hash algo
166170
_, _ = h.Write([]byte(hashKey))
167171
routine := h.Sum32() % tw.workerCount
172+
log.L(ctx).Debugf("Queuing write operation %s to worker tx_writer_%.4d", op.opID, routine)
168173
select {
169174
case tw.workQueues[routine] <- op: // it's queued
170175
case <-ctx.Done(): // timeout of caller context
@@ -180,6 +185,7 @@ func (tw *transactionWriter) worker(i int) {
180185
defer close(tw.workersDone[i])
181186
workerID := fmt.Sprintf("tx_writer_%.4d", i)
182187
ctx := log.WithLogField(tw.bgCtx, "job", workerID)
188+
l := log.L(ctx)
183189
var batch *transactionWriterBatch
184190
batchCount := 0
185191
workQueue := tw.workQueues[i]
@@ -202,24 +208,27 @@ func (tw *transactionWriter) worker(i int) {
202208
}
203209
if batch == nil {
204210
batch = &transactionWriterBatch{
205-
id: fmt.Sprintf("%.4d_%.9d", i, batchCount),
211+
id: fmt.Sprintf("%.4d_%.9d", i, batchCount),
212+
opened: time.Now(),
206213
}
207214
batch.timeoutContext, batch.timeoutCancel = context.WithTimeout(ctx, tw.batchTimeout)
208215
batchCount++
209216
}
210217
batch.ops = append(batch.ops, op)
218+
l.Debugf("Added write operation %s to batch %s (len=%d)", op.opID, batch.id, len(batch.ops))
211219
case <-timeoutContext.Done():
212220
timedOut = true
213221
select {
214222
case <-ctx.Done():
215-
log.L(ctx).Debugf("Transaction writer ending")
223+
l.Debugf("Transaction writer ending")
216224
return
217225
default:
218226
}
219227
}
220228

221229
if batch != nil && (timedOut || (len(batch.ops) >= tw.batchMaxSize)) {
222230
batch.timeoutCancel()
231+
l.Debugf("Running batch %s (len=%d,timeout=%t,age=%dms)", batch.id, len(batch.ops), timedOut, time.Since(batch.opened).Milliseconds())
223232
tw.runBatch(ctx, batch)
224233
batch = nil
225234
}
@@ -383,6 +392,7 @@ func (tw *transactionWriter) preInsertIdempotencyCheck(ctx context.Context, b *t
383392
txOp.sentConflict = true
384393
txOp.done <- i18n.NewError(ctx, tmmsgs.MsgDuplicateID, txOp.txID)
385394
} else {
395+
log.L(ctx).Debugf("Adding TX %s from write operation %s to insert idx=%d", txOp.txID, txOp.opID, len(validInserts))
386396
validInserts = append(validInserts, txOp.txInsert)
387397
}
388398
}
@@ -413,9 +423,19 @@ func (tw *transactionWriter) executeBatchOps(ctx context.Context, b *transaction
413423
}
414424
}
415425
// Do all the transaction updates
426+
mergedUpdates := make(map[string]*apitypes.TXUpdates)
416427
for _, op := range b.txUpdates {
417-
if err := tw.p.updateTransaction(ctx, op.txID, op.txUpdate); err != nil {
418-
log.L(ctx).Errorf("Update transaction %s failed: %s", op.txID, err)
428+
update, merge := mergedUpdates[op.txID]
429+
if merge {
430+
update.Merge(op.txUpdate)
431+
} else {
432+
mergedUpdates[op.txID] = op.txUpdate
433+
}
434+
log.L(ctx).Debugf("Updating transaction %s in write operation %s (merged=%t)", op.txID, op.opID, merge)
435+
}
436+
for txID, update := range mergedUpdates {
437+
if err := tw.p.updateTransaction(ctx, txID, update); err != nil {
438+
log.L(ctx).Errorf("Update transaction %s failed: %s", txID, err)
419439
return err
420440
}
421441
}

‎internal/persistence/postgres/transaction_writer_test.go

+73
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@ import (
2727
"github.com/hyperledger/firefly-common/pkg/fftypes"
2828
"github.com/hyperledger/firefly-transaction-manager/pkg/apitypes"
2929
"github.com/hyperledger/firefly-transaction-manager/pkg/ffcapi"
30+
"github.com/sirupsen/logrus"
3031
"github.com/stretchr/testify/assert"
3132
)
3233

@@ -275,6 +276,48 @@ func TestExecuteBatchOpsUpdateTXFail(t *testing.T) {
275276
assert.NoError(t, mdb.ExpectationsWereMet())
276277
}
277278

279+
func TestExecuteBatchOpsUpdateTXMerge(t *testing.T) {
280+
logrus.SetLevel(logrus.TraceLevel)
281+
282+
ctx, p, mdb, done := newMockSQLPersistence(t)
283+
defer done()
284+
285+
mdb.ExpectBegin()
286+
mdb.ExpectExec("UPDATE.*").WillReturnResult(sqlmock.NewResult(-1, 1))
287+
mdb.ExpectExec("UPDATE.*").WillReturnResult(sqlmock.NewResult(-1, 1))
288+
mdb.ExpectCommit()
289+
290+
err := p.db.RunAsGroup(ctx, func(ctx context.Context) error {
291+
return p.writer.executeBatchOps(ctx, &transactionWriterBatch{
292+
txUpdates: []*transactionOperation{
293+
{
294+
txID: "11111",
295+
txUpdate: &apitypes.TXUpdates{
296+
Status: ptrTo(apitypes.TxStatusPending),
297+
From: strPtr("0xaaaaa"),
298+
},
299+
},
300+
{
301+
txID: "22222",
302+
txUpdate: &apitypes.TXUpdates{
303+
Status: ptrTo(apitypes.TxStatusPending),
304+
},
305+
},
306+
{
307+
txID: "11111",
308+
txUpdate: &apitypes.TXUpdates{
309+
Status: ptrTo(apitypes.TxStatusSucceeded),
310+
TransactionHash: strPtr("0xaabbcc"),
311+
},
312+
},
313+
},
314+
})
315+
})
316+
assert.NoError(t, err)
317+
318+
assert.NoError(t, mdb.ExpectationsWereMet())
319+
}
320+
278321
func TestExecuteBatchOpsUpsertReceiptFail(t *testing.T) {
279322
ctx, p, mdb, done := newMockSQLPersistence(t)
280323
defer done()
@@ -455,3 +498,33 @@ func TestQueueClosedContext(t *testing.T) {
455498
p.writer.queue(closedCtx, newTransactionOperation("tx1"))
456499

457500
}
501+
502+
func TestStopDoneWorker(t *testing.T) {
503+
tw := &transactionWriter{
504+
workersDone: []chan struct{}{
505+
make(chan struct{}),
506+
},
507+
}
508+
tw.bgCtx, tw.cancelCtx = context.WithCancel(context.Background())
509+
close(tw.workersDone[0])
510+
tw.stop()
511+
}
512+
513+
func TestStopDoneCtx(t *testing.T) {
514+
tw := &transactionWriter{
515+
workersDone: []chan struct{}{
516+
make(chan struct{}, 1),
517+
},
518+
}
519+
tw.bgCtx, tw.cancelCtx = context.WithCancel(context.Background())
520+
tw.cancelCtx()
521+
go func() {
522+
time.Sleep(10 * time.Millisecond)
523+
tw.workersDone[0] <- struct{}{}
524+
}()
525+
tw.stop()
526+
}
527+
528+
func ptrTo[T any](v T) *T {
529+
return &v
530+
}

‎pkg/apitypes/managed_tx.go

+46-1
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
// Copyright © 2023 Kaleido, Inc.
1+
// Copyright © 2024 Kaleido, Inc.
22
//
33
// SPDX-License-Identifier: Apache-2.0
44
//
@@ -208,6 +208,51 @@ type TXUpdates struct {
208208
ErrorMessage *string `json:"errorMessage,omitempty"`
209209
}
210210

211+
func (txu *TXUpdates) Merge(txu2 *TXUpdates) {
212+
if txu2.Status != nil {
213+
txu.Status = txu2.Status
214+
}
215+
if txu2.DeleteRequested != nil {
216+
txu.DeleteRequested = txu2.DeleteRequested
217+
}
218+
if txu2.From != nil {
219+
txu.From = txu2.From
220+
}
221+
if txu2.To != nil {
222+
txu.To = txu2.To
223+
}
224+
if txu2.Nonce != nil {
225+
txu.Nonce = txu2.Nonce
226+
}
227+
if txu2.Gas != nil {
228+
txu.Gas = txu2.Gas
229+
}
230+
if txu2.Value != nil {
231+
txu.Value = txu2.Value
232+
}
233+
if txu2.GasPrice != nil {
234+
txu.GasPrice = txu2.GasPrice
235+
}
236+
if txu2.TransactionData != nil {
237+
txu.TransactionData = txu2.TransactionData
238+
}
239+
if txu2.TransactionHash != nil {
240+
txu.TransactionHash = txu2.TransactionHash
241+
}
242+
if txu2.PolicyInfo != nil {
243+
txu.PolicyInfo = txu2.PolicyInfo
244+
}
245+
if txu2.FirstSubmit != nil {
246+
txu.FirstSubmit = txu2.FirstSubmit
247+
}
248+
if txu2.LastSubmit != nil {
249+
txu.LastSubmit = txu2.LastSubmit
250+
}
251+
if txu2.ErrorMessage != nil {
252+
txu.ErrorMessage = txu2.ErrorMessage
253+
}
254+
}
255+
211256
// TXWithStatus is a convenience object that fetches all data about a transaction into one
212257
// large JSON payload (with limits on certain parts, such as the history entries).
213258
// Note that in LevelDB persistence this is the stored form of the single document object.

‎pkg/apitypes/managed_tx_test.go

+28
Original file line numberDiff line numberDiff line change
@@ -85,3 +85,31 @@ func TestReceiptRecord(t *testing.T) {
8585
r.SetUpdated(t2)
8686
assert.Equal(t, t2, r.Updated)
8787
}
88+
89+
func TestTXUpdatesMerge(t *testing.T) {
90+
txu := &TXUpdates{}
91+
txu2 := &TXUpdates{
92+
Status: ptrTo(TxStatusPending),
93+
DeleteRequested: fftypes.Now(),
94+
From: ptrTo("1111"),
95+
To: ptrTo("2222"),
96+
Nonce: fftypes.NewFFBigInt(3333),
97+
Gas: fftypes.NewFFBigInt(4444),
98+
Value: fftypes.NewFFBigInt(5555),
99+
GasPrice: fftypes.JSONAnyPtr(`{"some": "stuff"}`),
100+
TransactionData: ptrTo("xxxx"),
101+
TransactionHash: ptrTo("yyyy"),
102+
PolicyInfo: fftypes.JSONAnyPtr(`{"more": "stuff"}`),
103+
FirstSubmit: fftypes.Now(),
104+
LastSubmit: fftypes.Now(),
105+
ErrorMessage: ptrTo("pop"),
106+
}
107+
txu.Merge(txu2)
108+
assert.Equal(t, *txu2, *txu)
109+
txu.Merge(&TXUpdates{})
110+
assert.Equal(t, *txu2, *txu)
111+
}
112+
113+
func ptrTo[T any](v T) *T {
114+
return &v
115+
}

‎pkg/apitypes/query_request.go

+2-1
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
// Copyright © 2022 Kaleido, Inc.
1+
// Copyright © 2024 Kaleido, Inc.
22
//
33
// SPDX-License-Identifier: Apache-2.0
44
//
@@ -24,6 +24,7 @@ import (
2424
type QueryRequest struct {
2525
Headers RequestHeaders `json:"headers"`
2626
ffcapi.TransactionInput
27+
BlockNumber *string `json:"blockNumber,omitempty"`
2728
}
2829

2930
// QueryResponse is the response payload for a query

‎pkg/ffcapi/method_call.go

+2-2
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
// Copyright © 2022 Kaleido, Inc.
1+
// Copyright © 2024 Kaleido, Inc.
22
//
33
// SPDX-License-Identifier: Apache-2.0
44
//
@@ -28,7 +28,7 @@ import (
2828
// detected by the back-end connector.
2929
type QueryInvokeRequest struct {
3030
TransactionInput
31-
BlockNumber *fftypes.FFBigInt `json:"blockNumber,omitempty"`
31+
BlockNumber *string `json:"blockNumber,omitempty"`
3232
}
3333

3434
type QueryInvokeResponse struct {

‎pkg/fftm/route__root_command.go

+2-1
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
// Copyright © 2023 Kaleido, Inc.
1+
// Copyright © 2024 Kaleido, Inc.
22
//
33
// SPDX-License-Identifier: Apache-2.0
44
//
@@ -101,6 +101,7 @@ var postRootCommand = func(m *manager) *ffapi.Route {
101101
}
102102
res, _, err := m.connector.QueryInvoke(r.Req.Context(), &ffcapi.QueryInvokeRequest{
103103
TransactionInput: tReq.TransactionInput,
104+
BlockNumber: tReq.BlockNumber,
104105
})
105106
if err != nil {
106107
return nil, err

0 commit comments

Comments
 (0)