1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
|
package txinfo
import (
"context"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"gitlab.com/gitlab-org/gitaly/internal/backchannel"
"google.golang.org/grpc/metadata"
)
const (
// TransactionMetadataKey is the key used to store transaction
// information in the gRPC metadata.
TransactionMetadataKey = "gitaly-reference-transaction"
)
var (
// ErrTransactionNotFound indicates the transaction metadata could not
// be found
ErrTransactionNotFound = errors.New("transaction not found")
)
// Transaction stores parameters required to identify a reference
// transaction.
type Transaction struct {
// BackchannelID is the ID of the backchannel that corresponds to the Praefect
// that is handling the transaction. This field is filled in by the Gitaly.
BackchannelID backchannel.ID `json:"backchannel_id,omitempty"`
// ID is the unique identifier of a transaction
ID uint64 `json:"id"`
// Node is the name used to cast a vote
Node string `json:"node"`
// Primary identifies the node's role in this transaction
Primary bool `json:"primary"`
}
// serialize serializes a `Transaction` into a string.
func (t Transaction) serialize() (string, error) {
marshalled, err := json.Marshal(t)
if err != nil {
return "", err
}
return base64.StdEncoding.EncodeToString(marshalled), nil
}
// transactionFromSerialized creates a transaction from a `serialize()`d string.
func transactionFromSerialized(serialized string) (Transaction, error) {
decoded, err := base64.StdEncoding.DecodeString(serialized)
if err != nil {
return Transaction{}, err
}
var tx Transaction
if err := json.Unmarshal(decoded, &tx); err != nil {
return Transaction{}, err
}
return tx, nil
}
// InjectTransaction injects reference transaction metadata into an incoming context
func InjectTransaction(ctx context.Context, tranasctionID uint64, node string, primary bool) (context.Context, error) {
transaction := Transaction{
ID: tranasctionID,
Node: node,
Primary: primary,
}
serialized, err := transaction.serialize()
if err != nil {
return nil, err
}
md, ok := metadata.FromIncomingContext(ctx)
if !ok {
md = metadata.New(map[string]string{})
} else {
md = md.Copy()
}
md.Set(TransactionMetadataKey, serialized)
return metadata.NewIncomingContext(ctx, md), nil
}
// TransactionFromContext extracts `Transaction` from an incoming context. In
// case the metadata key is not set, the function will return
// `ErrTransactionNotFound`.
func TransactionFromContext(ctx context.Context) (Transaction, error) {
md, ok := metadata.FromIncomingContext(ctx)
if !ok {
return Transaction{}, ErrTransactionNotFound
}
serialized := md[TransactionMetadataKey]
if len(serialized) == 0 {
return Transaction{}, ErrTransactionNotFound
}
transaction, err := transactionFromSerialized(serialized[0])
if err != nil {
return Transaction{}, fmt.Errorf("from serialized: %w", err)
}
// For backwards compatibility during an upgrade, we still need to accept transactions
// from non-multiplexed connections. From 14.0 onwards, we can expect every transaction to
// originate from a multiplexed connection and should drop the error check below.
transaction.BackchannelID, err = backchannel.GetPeerID(ctx)
if err != nil && !errors.Is(err, backchannel.ErrNonMultiplexedConnection) {
return Transaction{}, fmt.Errorf("get peer id: %w", err)
}
return transaction, nil
}
// FromContext extracts transaction-related metadata from the given context. No error is returned in
// case no transaction was found.
func FromContext(ctx context.Context) (*Transaction, *PraefectServer, error) {
transaction, err := TransactionFromContext(ctx)
if err != nil {
if err != ErrTransactionNotFound {
return nil, nil, err
}
return nil, nil, nil
}
praefect, err := PraefectFromContext(ctx)
if err != nil {
return nil, nil, err
}
return &transaction, praefect, nil
}
|
