Skip to content

ttx service: timeout via context - part 1 #1028

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 7 commits into from
Apr 16, 2025
Merged

ttx service: timeout via context - part 1 #1028

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
dc97d48
issue fix
sreenidhidrIBM Mar 21, 2025
25ccfdd
err handling
sreenidhidrIBM Mar 27, 2025
789fb9e
err handling
sreenidhidrIBM Mar 27, 2025
d0b01a1
read message func removal
sreenidhidrIBM Apr 1, 2025
c13d2bb
read message func addition
sreenidhidrIBM Apr 1, 2025
161b9e4
read message func deletion
sreenidhidrIBM Apr 1, 2025
ff29240
review comment
sreenidhidrIBM Apr 11, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 3 additions & 7 deletions token/services/ttx/accept.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ import (
"github.com/hyperledger-labs/fabric-smart-client/platform/view/view"
"github.com/hyperledger-labs/fabric-token-sdk/token"
"github.com/hyperledger-labs/fabric-token-sdk/token/services/tokens"
session2 "github.com/hyperledger-labs/fabric-token-sdk/token/services/utils/json/session"
"github.com/pkg/errors"
"go.opentelemetry.io/otel/trace"
"go.uber.org/zap/zapcore"
Expand Down Expand Up @@ -132,16 +133,11 @@ func (s *AcceptView) respondToSignatureRequests(context view.Context) error {
if logger.IsEnabledFor(zapcore.DebugLevel) {
logger.Debugf("Receiving signature request...")
}

msg, err := ReadMessage(session, time.Minute)
jsonSession := session2.JSON(context)
err := jsonSession.ReceiveWithTimeout(signatureRequest, time.Minute)
if err != nil {
return errors.Wrap(err, "failed reading signature request")
}
// TODO: check what is signed...
err = Unmarshal(msg, signatureRequest)
if err != nil {
return errors.Wrap(err, "failed unmarshalling signature request")
}
}
span.AddEvent("Fetched request from session")
tms := token.GetManagementService(context, token.WithTMS(s.tx.Network(), s.tx.Channel(), s.tx.Namespace()))
Expand Down
4 changes: 3 additions & 1 deletion token/services/ttx/auditor.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ import (
"github.com/hyperledger-labs/fabric-token-sdk/token/services/db/driver"
"github.com/hyperledger-labs/fabric-token-sdk/token/services/tokens"
"github.com/hyperledger-labs/fabric-token-sdk/token/services/ttxdb"
session2 "github.com/hyperledger-labs/fabric-token-sdk/token/services/utils/json/session"
view3 "github.com/hyperledger-labs/fabric-token-sdk/token/services/utils/view"
"github.com/pkg/errors"
"go.opentelemetry.io/otel/trace"
Expand Down Expand Up @@ -146,7 +147,8 @@ func (a *AuditingViewInitiator) Call(context view.Context) (interface{}, error)
// Receive signature
logger.Debugf("Receiving signature for [%s]", a.tx.ID())
span.AddEvent("start_receiving")
signature, err := ReadMessage(session, time.Minute)
jsonSession := session2.NewFromSession(context, session)
signature, err := jsonSession.ReceiveRawWithTimeout(time.Minute)
if err != nil {
span.RecordError(err)
return nil, errors.WithMessage(err, "failed to read audit event")
Expand Down
15 changes: 9 additions & 6 deletions token/services/ttx/endorse.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ import (
"github.com/hyperledger-labs/fabric-token-sdk/token/services/identity/multisig"
"github.com/hyperledger-labs/fabric-token-sdk/token/services/network"
"github.com/hyperledger-labs/fabric-token-sdk/token/services/tokens"
session2 "github.com/hyperledger-labs/fabric-token-sdk/token/services/utils/json/session"
"github.com/pkg/errors"
"go.opentelemetry.io/otel/trace"
"go.uber.org/zap/zapcore"
Expand Down Expand Up @@ -349,11 +350,11 @@ func (c *CollectEndorsementsView) signRemote(context view.Context, party view.Id
return nil, errors.Wrap(err, "failed sending transaction content")
}

sigma, err := ReadMessage(session, time.Minute)
jsonSession := session2.NewFromSession(context, session)
sigma, err := jsonSession.ReceiveRawWithTimeout(time.Minute)
if err != nil {
return nil, errors.Wrap(err, "failed reading message")
}

verifier, err := verifierGetter(party)
if err != nil {
return nil, errors.Wrapf(err, "failed getting verifier for [%s]", party)
Expand Down Expand Up @@ -554,7 +555,8 @@ func (c *CollectEndorsementsView) distributeEvnToParty(context view.Context, ent
}

span.AddEvent("Wait for ack")
sigma, err := ReadMessage(session, 1*time.Minute)
jsonSession := session2.NewFromSession(context, session)
sigma, err := jsonSession.ReceiveRawWithTimeout(time.Minute)
if err != nil {
return errors.Wrapf(err, "failed reading message on session [%s]", session.Info().ID)
}
Expand Down Expand Up @@ -736,8 +738,8 @@ func (f *ReceiveTransactionView) Call(context view.Context) (interface{}, error)
span := trace.SpanFromContext(context.Context())
span.AddEvent("start_receive_transaction_view")
defer span.AddEvent("end_receive_transaction_view")

msg, err := ReadMessage(context.Session(), time.Minute*4)
jsonSession := session2.JSON(context)
msg, err := jsonSession.ReceiveRawWithTimeout(time.Minute * 4)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we could also extract these durations to constants, depending on what we are waiting for, e.g. signatureRcvTimeout, txRcvTimeout, to make sure we are aligned and to have a better overview. ideally I would even add them as fields in the structure objects and set the default values in the constructors.

if err != nil {
span.RecordError(err)
}
Expand Down Expand Up @@ -845,7 +847,8 @@ func (s *EndorseView) Call(context view.Context) (interface{}, error) {
if logger.IsEnabledFor(zapcore.DebugLevel) {
logger.Debugf("Receiving signature request...")
}
srRaw, err = ReadMessage(session, time.Minute)
jsonSession := session2.JSON(context)
srRaw, err = jsonSession.ReceiveRawWithTimeout(time.Minute)
if err != nil {
return nil, errors.Wrap(err, "failed reading signature request")
}
Expand Down
30 changes: 3 additions & 27 deletions token/services/ttx/multisig/spend.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,12 +7,10 @@ SPDX-License-Identifier: Apache-2.0
package multisig

import (
"runtime/debug"
"slices"
"time"

"github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors"
"github.com/hyperledger-labs/fabric-smart-client/platform/view/services/hash"
"github.com/hyperledger-labs/fabric-smart-client/platform/view/view"
token2 "github.com/hyperledger-labs/fabric-token-sdk/token"
"github.com/hyperledger-labs/fabric-token-sdk/token/core/common/encoding/json"
Expand All @@ -21,23 +19,13 @@ import (
"github.com/hyperledger-labs/fabric-token-sdk/token/services/utils/json/session"
"github.com/hyperledger-labs/fabric-token-sdk/token/token"
"go.opentelemetry.io/otel/trace"
"go.uber.org/zap/zapcore"
)

// SpendRequest is the request to spend a token
type SpendRequest struct {
Token *token.UnspentToken
}

func NewSpendRequestFromBytes(msg []byte) (*SpendRequest, error) {
request := &SpendRequest{}
err := json.Unmarshal(msg, request)
if err != nil {
return nil, errors.Wrap(err, "failed unmarshalling spendRequest")
}
return request, nil
}

func ReceiveSpendRequest(context view.Context, opts ...ttx.TxOption) (*SpendRequest, error) {
logger.Debugf("receive a new spendRequest...")
requestBoxed, err := context.RunView(NewReceiveSpendRequestView(), view.WithSameContext())
Expand Down Expand Up @@ -66,25 +54,13 @@ func (f *ReceiveSpendRequestView) Call(context view.Context) (interface{}, error
span := trace.SpanFromContext(context.Context())
span.AddEvent("start_receive_spendRequest_view")
defer span.AddEvent("end_receive_spendRequest_view")

msg, err := ttx.ReadMessage(context.Session(), time.Minute*4)
tx := &SpendRequest{}
jsonSession := session.JSON(context)
err := jsonSession.ReceiveWithTimeout(tx, time.Minute*4)
if err != nil {
span.RecordError(err)
}
span.AddEvent("receive_tx")

if logger.IsEnabledFor(zapcore.DebugLevel) {
logger.Debugf("ReceiveSpendRequestView: received spendRequest, len [%d][%s]", len(msg), hash.Hashable(msg))
}
if len(msg) == 0 {
info := context.Session().Info()
logger.Errorf("received empty message, session closed [%s:%v], [%s]", info.ID, info.Closed, string(debug.Stack()))
return nil, errors.Errorf("received empty message, session closed [%s:%v]", info.ID, info.Closed)
}
tx, err := NewSpendRequestFromBytes(msg)
if err != nil {
return nil, errors.Wrap(err, "failed to receive spendRequest")
}
return tx, nil
}

Expand Down
21 changes: 0 additions & 21 deletions token/services/ttx/session.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,6 @@ package ttx
import (
"context"
"encoding/base64"
"time"

"github.com/hyperledger-labs/fabric-smart-client/platform/view/view"
"github.com/pkg/errors"
Expand Down Expand Up @@ -128,23 +127,3 @@ func (s *localSession) Receive() <-chan *view.Message {
func (s *localSession) Close() {
s.info.Closed = true
}

func ReadMessage(session view.Session, timeout time.Duration) ([]byte, error) {
timer := time.NewTimer(timeout)
defer timer.Stop()

ch := session.Receive()
select {
case msg := <-ch:
if msg == nil {
return nil, errors.New("received nil tx")
}
if msg.Status == view.ERROR {
return nil, errors.New(string(msg.Payload))
}
return msg.Payload, nil
case <-timer.C:
err := errors.New("timeout reached")
return nil, err
}
}