some more fixes
This commit is contained in:
@ -7,6 +7,7 @@ import (
|
||||
"os"
|
||||
|
||||
pb "git.dws.rip/dubey/kat/api/v1alpha1"
|
||||
"github.com/davecgh/go-spew/spew"
|
||||
"gopkg.in/yaml.v3"
|
||||
|
||||
"encoding/json"
|
||||
@ -64,6 +65,8 @@ func ParseClusterConfiguration(filePath string) (*pb.ClusterConfiguration, error
|
||||
return nil, fmt.Errorf("failed to unmarshal spec into proto: %w", err)
|
||||
}
|
||||
|
||||
spew.Dump(&config) // For debugging, remove in production
|
||||
|
||||
SetClusterConfigDefaults(&config)
|
||||
|
||||
if err := ValidateClusterConfiguration(&config); err != nil {
|
||||
|
||||
464
internal/store/etcd.go
Normal file
464
internal/store/etcd.go
Normal file
@ -0,0 +1,464 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log"
|
||||
"net/url"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"go.etcd.io/etcd/client/v3/concurrency"
|
||||
"go.etcd.io/etcd/server/v3/embed"
|
||||
"go.etcd.io/etcd/server/v3/etcdserver/api/v3client"
|
||||
|
||||
clientv3 "go.etcd.io/etcd/client/v3"
|
||||
)
|
||||
|
||||
const (
|
||||
defaultDialTimeout = 5 * time.Second
|
||||
defaultRequestTimeout = 5 * time.Second
|
||||
leaderElectionPrefix = "/kat/leader_election/"
|
||||
)
|
||||
|
||||
// EtcdEmbedConfig holds configuration for an embedded etcd server.
|
||||
type EtcdEmbedConfig struct {
|
||||
Name string
|
||||
DataDir string
|
||||
ClientURLs []string // URLs for client communication
|
||||
PeerURLs []string // URLs for peer communication
|
||||
InitialCluster string // e.g., "node1=http://localhost:2380"
|
||||
// Add other etcd config fields as needed: LogLevel, etc.
|
||||
}
|
||||
|
||||
// EtcdStore implements the StateStore interface using etcd.
|
||||
type EtcdStore struct {
|
||||
client *clientv3.Client
|
||||
etcdServer *embed.Etcd // Holds the embedded server instance, if any
|
||||
|
||||
// For leadership
|
||||
session *concurrency.Session
|
||||
election *concurrency.Election
|
||||
leaderID string
|
||||
leaseTTL int64
|
||||
campaignCtx context.Context
|
||||
campaignDone func() // Cancels campaignCtx
|
||||
resignMutex sync.Mutex // Protects session and election during resign
|
||||
}
|
||||
|
||||
// StartEmbeddedEtcd starts an embedded etcd server based on the provided config.
|
||||
func StartEmbeddedEtcd(cfg EtcdEmbedConfig) (*embed.Etcd, error) {
|
||||
embedCfg := embed.NewConfig()
|
||||
embedCfg.Name = cfg.Name
|
||||
embedCfg.Dir = cfg.DataDir
|
||||
embedCfg.InitialClusterToken = "kat-etcd-cluster" // Make this configurable if needed
|
||||
embedCfg.InitialCluster = cfg.InitialCluster
|
||||
embedCfg.ForceNewCluster = false // Set to true only for initial bootstrap of a new cluster if needed
|
||||
|
||||
lpurl, err := parseURLs(cfg.PeerURLs)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("invalid peer URLs: %w", err)
|
||||
}
|
||||
embedCfg.ListenPeerUrls = lpurl
|
||||
|
||||
lcurl, err := parseURLs(cfg.ClientURLs)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("invalid client URLs: %w", err)
|
||||
}
|
||||
embedCfg.ListenClientUrls = lcurl
|
||||
|
||||
// TODO: Configure logging, metrics, etc. for embedded etcd
|
||||
// embedCfg.Logger = "zap"
|
||||
// embedCfg.LogLevel = "info"
|
||||
|
||||
e, err := embed.StartEtcd(embedCfg)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to start embedded etcd: %w", err)
|
||||
}
|
||||
|
||||
select {
|
||||
case <-e.Server.ReadyNotify():
|
||||
log.Printf("Embedded etcd server is ready (name: %s)", cfg.Name)
|
||||
case <-time.After(60 * time.Second): // Adjust timeout as needed
|
||||
e.Server.Stop() // trigger a shutdown
|
||||
return nil, fmt.Errorf("embedded etcd server took too long to start")
|
||||
}
|
||||
return e, nil
|
||||
}
|
||||
|
||||
func parseURLs(urlsStr []string) ([]url.URL, error) {
|
||||
urls := make([]url.URL, len(urlsStr))
|
||||
for i, s := range urlsStr {
|
||||
u, err := url.Parse(s)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("parsing URL '%s': %w", s, err)
|
||||
}
|
||||
urls[i] = *u
|
||||
}
|
||||
return urls, nil
|
||||
}
|
||||
|
||||
// NewEtcdStore creates a new EtcdStore.
|
||||
// If etcdServer is not nil, it assumes it's managing an embedded server.
|
||||
// endpoints are the etcd client endpoints.
|
||||
func NewEtcdStore(endpoints []string, etcdServer *embed.Etcd) (*EtcdStore, error) {
|
||||
var cli *clientv3.Client
|
||||
var err error
|
||||
|
||||
if etcdServer != nil {
|
||||
// If embedded server is provided, use its client directly
|
||||
cli = v3client.New(etcdServer.Server)
|
||||
} else {
|
||||
cli, err = clientv3.New(clientv3.Config{
|
||||
Endpoints: endpoints,
|
||||
DialTimeout: defaultDialTimeout,
|
||||
// TODO: Add TLS config if connecting to secure external etcd
|
||||
})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to create etcd client: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
return &EtcdStore{
|
||||
client: cli,
|
||||
etcdServer: etcdServer,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *EtcdStore) Put(ctx context.Context, key string, value []byte) error {
|
||||
reqCtx, cancel := context.WithTimeout(ctx, defaultRequestTimeout)
|
||||
defer cancel()
|
||||
_, err := s.client.Put(reqCtx, key, string(value))
|
||||
return err
|
||||
}
|
||||
|
||||
func (s *EtcdStore) Get(ctx context.Context, key string) (*KV, error) {
|
||||
reqCtx, cancel := context.WithTimeout(ctx, defaultRequestTimeout)
|
||||
defer cancel()
|
||||
resp, err := s.client.Get(reqCtx, key)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(resp.Kvs) == 0 {
|
||||
return nil, fmt.Errorf("key not found: %s", key) // Or a specific error type
|
||||
}
|
||||
kv := resp.Kvs[0]
|
||||
return &KV{
|
||||
Key: string(kv.Key),
|
||||
Value: kv.Value,
|
||||
Version: kv.ModRevision,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *EtcdStore) Delete(ctx context.Context, key string) error {
|
||||
reqCtx, cancel := context.WithTimeout(ctx, defaultRequestTimeout)
|
||||
defer cancel()
|
||||
_, err := s.client.Delete(reqCtx, key)
|
||||
return err
|
||||
}
|
||||
|
||||
func (s *EtcdStore) List(ctx context.Context, prefix string) ([]KV, error) {
|
||||
reqCtx, cancel := context.WithTimeout(ctx, defaultRequestTimeout)
|
||||
defer cancel()
|
||||
resp, err := s.client.Get(reqCtx, prefix, clientv3.WithPrefix())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
kvs := make([]KV, len(resp.Kvs))
|
||||
for i, etcdKv := range resp.Kvs {
|
||||
kvs[i] = KV{
|
||||
Key: string(etcdKv.Key),
|
||||
Value: etcdKv.Value,
|
||||
Version: etcdKv.ModRevision,
|
||||
}
|
||||
}
|
||||
return kvs, nil
|
||||
}
|
||||
|
||||
func (s *EtcdStore) Watch(ctx context.Context, keyOrPrefix string, startRevision int64) (<-chan WatchEvent, error) {
|
||||
watchChan := make(chan WatchEvent)
|
||||
opts := []clientv3.OpOption{clientv3.WithPrefix()}
|
||||
if startRevision > 0 {
|
||||
opts = append(opts, clientv3.WithRev(startRevision))
|
||||
}
|
||||
|
||||
etcdWatchChan := s.client.Watch(ctx, keyOrPrefix, opts...)
|
||||
|
||||
go func() {
|
||||
defer close(watchChan)
|
||||
for resp := range etcdWatchChan {
|
||||
if err := resp.Err(); err != nil {
|
||||
log.Printf("EtcdStore watch error: %v", err)
|
||||
// Depending on error, might need to signal channel consumer
|
||||
return
|
||||
}
|
||||
for _, ev := range resp.Events {
|
||||
event := WatchEvent{
|
||||
KV: KV{
|
||||
Key: string(ev.Kv.Key),
|
||||
Value: ev.Kv.Value,
|
||||
Version: ev.Kv.ModRevision,
|
||||
},
|
||||
}
|
||||
if ev.PrevKv != nil {
|
||||
event.PrevKV = &KV{
|
||||
Key: string(ev.PrevKv.Key),
|
||||
Value: ev.PrevKv.Value,
|
||||
Version: ev.PrevKv.ModRevision,
|
||||
}
|
||||
}
|
||||
|
||||
switch ev.Type {
|
||||
case clientv3.EventTypePut:
|
||||
event.Type = EventTypePut
|
||||
case clientv3.EventTypeDelete:
|
||||
event.Type = EventTypeDelete
|
||||
default:
|
||||
log.Printf("EtcdStore unknown event type: %v", ev.Type)
|
||||
continue
|
||||
}
|
||||
select {
|
||||
case watchChan <- event:
|
||||
case <-ctx.Done():
|
||||
log.Printf("EtcdStore watch context cancelled for %s", keyOrPrefix)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
return watchChan, nil
|
||||
}
|
||||
|
||||
func (s *EtcdStore) Close() error {
|
||||
s.resignMutex.Lock()
|
||||
if s.session != nil {
|
||||
// Attempt to close session gracefully, which should also resign from election
|
||||
// if campaign was active.
|
||||
s.session.Close() // This is synchronous
|
||||
s.session = nil
|
||||
s.election = nil
|
||||
if s.campaignDone != nil {
|
||||
s.campaignDone() // Ensure leadership context is cancelled
|
||||
s.campaignDone = nil
|
||||
}
|
||||
}
|
||||
s.resignMutex.Unlock()
|
||||
|
||||
var clientErr error
|
||||
if s.client != nil {
|
||||
clientErr = s.client.Close()
|
||||
}
|
||||
if s.etcdServer != nil {
|
||||
s.etcdServer.Close() // This stops the embedded server
|
||||
}
|
||||
|
||||
if clientErr != nil {
|
||||
return fmt.Errorf("error closing etcd client: %w", clientErr)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *EtcdStore) Campaign(ctx context.Context, leaderID string, leaseTTLSeconds int64) (leadershipCtx context.Context, err error) {
|
||||
s.resignMutex.Lock()
|
||||
defer s.resignMutex.Unlock()
|
||||
|
||||
if s.session != nil {
|
||||
return nil, fmt.Errorf("campaign already in progress or session active")
|
||||
}
|
||||
|
||||
s.leaderID = leaderID
|
||||
s.leaseTTL = leaseTTLSeconds
|
||||
|
||||
// Create a new session
|
||||
session, err := concurrency.NewSession(s.client, concurrency.WithTTL(int(leaseTTLSeconds)))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to create etcd session: %w", err)
|
||||
}
|
||||
s.session = session
|
||||
|
||||
election := concurrency.NewElection(session, leaderElectionPrefix)
|
||||
s.election = election
|
||||
|
||||
// Create a cancellable context for this campaign attempt
|
||||
// This context will be returned and is cancelled when leadership is lost or Resign is called.
|
||||
campaignSpecificCtx, cancelCampaignSpecificCtx := context.WithCancel(ctx)
|
||||
s.campaignCtx = campaignSpecificCtx
|
||||
s.campaignDone = cancelCampaignSpecificCtx
|
||||
|
||||
go func() {
|
||||
defer func() {
|
||||
// This block ensures that if the campaign goroutine exits for any reason
|
||||
// (e.g. session.Done(), campaign error, context cancellation),
|
||||
// the leadership context is cancelled.
|
||||
s.resignMutex.Lock()
|
||||
if s.campaignDone != nil { // Check if not already resigned
|
||||
s.campaignDone()
|
||||
s.campaignDone = nil // Prevent double cancel
|
||||
}
|
||||
// Clean up session if it's still this one
|
||||
if s.session == session {
|
||||
s.session.Close() // Attempt to close the session
|
||||
s.session = nil
|
||||
s.election = nil
|
||||
}
|
||||
s.resignMutex.Unlock()
|
||||
}()
|
||||
|
||||
// Campaign for leadership in a blocking way
|
||||
// The campaignCtx (parent context) can cancel this.
|
||||
if err := election.Campaign(s.campaignCtx, leaderID); err != nil {
|
||||
log.Printf("Error during leadership campaign for %s: %v", leaderID, err)
|
||||
// Error here usually means context cancelled or session closed.
|
||||
return
|
||||
}
|
||||
|
||||
// If Campaign returns without error, it means we are elected.
|
||||
// Keep leadership context alive until session is done or campaignCtx is cancelled.
|
||||
log.Printf("Successfully campaigned, %s is now leader", leaderID)
|
||||
|
||||
// Monitor the session; if it closes, leadership is lost.
|
||||
select {
|
||||
case <-session.Done():
|
||||
log.Printf("Etcd session closed for leader %s, leadership lost", leaderID)
|
||||
case <-s.campaignCtx.Done(): // This is campaignSpecificCtx
|
||||
log.Printf("Leadership campaign context cancelled for %s", leaderID)
|
||||
}
|
||||
}()
|
||||
|
||||
return s.campaignCtx, nil
|
||||
}
|
||||
|
||||
func (s *EtcdStore) Resign(ctx context.Context) error {
|
||||
s.resignMutex.Lock()
|
||||
defer s.resignMutex.Unlock()
|
||||
|
||||
if s.election == nil || s.session == nil {
|
||||
log.Println("Resign called but not currently leading or no active session.")
|
||||
return nil // Not an error to resign if not leading
|
||||
}
|
||||
|
||||
log.Printf("Resigning leadership for %s", s.leaderID)
|
||||
|
||||
// Cancel the leadership context
|
||||
if s.campaignDone != nil {
|
||||
s.campaignDone()
|
||||
s.campaignDone = nil
|
||||
}
|
||||
|
||||
// Resign from the election. This is a best-effort.
|
||||
// The context passed to Resign should be short-lived.
|
||||
resignCtx, cancel := context.WithTimeout(context.Background(), defaultRequestTimeout)
|
||||
defer cancel()
|
||||
if err := s.election.Resign(resignCtx); err != nil {
|
||||
log.Printf("Error resigning from election: %v. Session will eventually expire.", err)
|
||||
// Don't return error here, as session closure will handle it.
|
||||
}
|
||||
|
||||
// Close the session to ensure lease is revoked quickly.
|
||||
if s.session != nil {
|
||||
err := s.session.Close() // This is synchronous
|
||||
s.session = nil
|
||||
s.election = nil
|
||||
if err != nil {
|
||||
return fmt.Errorf("error closing session during resign: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
log.Printf("Successfully resigned leadership for %s", s.leaderID)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *EtcdStore) GetLeader(ctx context.Context) (string, error) {
|
||||
// This method needs a temporary session if one doesn't exist,
|
||||
// or it can try to get the leader key directly if the election pattern stores it.
|
||||
// concurrency.NewElection().Leader(ctx) is the way.
|
||||
// It requires a session. If we are campaigning, we have one.
|
||||
// If we are just an observer, we might need a short-lived session.
|
||||
|
||||
s.resignMutex.Lock()
|
||||
currentSession := s.session
|
||||
s.resignMutex.Unlock()
|
||||
|
||||
var tempSession *concurrency.Session
|
||||
var err error
|
||||
|
||||
if currentSession == nil {
|
||||
// Create a temporary session to observe leader
|
||||
// Use a shorter TTL for observer session if desired, or same as campaign TTL
|
||||
ttl := s.leaseTTL
|
||||
if ttl == 0 {
|
||||
ttl = 10 // Default observer TTL
|
||||
}
|
||||
tempSession, err = concurrency.NewSession(s.client, concurrency.WithTTL(int(ttl)))
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("failed to create temporary session for GetLeader: %w", err)
|
||||
}
|
||||
defer tempSession.Close()
|
||||
currentSession = tempSession
|
||||
}
|
||||
|
||||
election := concurrency.NewElection(currentSession, leaderElectionPrefix)
|
||||
reqCtx, cancel := context.WithTimeout(ctx, defaultRequestTimeout)
|
||||
defer cancel()
|
||||
|
||||
resp, err := election.Leader(reqCtx)
|
||||
if err != nil {
|
||||
if err == concurrency.ErrElectionNoLeader {
|
||||
return "", nil // No leader currently elected
|
||||
}
|
||||
return "", fmt.Errorf("failed to get leader: %w", err)
|
||||
}
|
||||
if resp != nil && len(resp.Kvs) > 0 {
|
||||
return string(resp.Kvs[0].Value), nil
|
||||
}
|
||||
return "", nil // No leader
|
||||
}
|
||||
|
||||
func (s *EtcdStore) DoTransaction(ctx context.Context, checks []Compare, onSuccess []Op, onFailure []Op) (bool, error) {
|
||||
if len(onFailure) > 0 {
|
||||
// Standard etcd Txn doesn't have an "Else" block that takes arbitrary operations
|
||||
// like K8s apiserver. It only has If/Then.
|
||||
// We can simulate simple Else cases if they are just Get ops, but not Puts/Deletes.
|
||||
// For now, let's state this limitation.
|
||||
return false, fmt.Errorf("onFailure operations are not fully supported in etcd transaction implementation")
|
||||
}
|
||||
|
||||
etcdCmps := make([]clientv3.Cmp, len(checks))
|
||||
for i, c := range checks {
|
||||
if c.ExpectedVersion == 0 { // Key should not exist
|
||||
etcdCmps[i] = clientv3.Compare(clientv3.ModRevision(c.Key), "=", 0)
|
||||
} else { // Key should exist with specific version
|
||||
etcdCmps[i] = clientv3.Compare(clientv3.ModRevision(c.Key), "=", c.ExpectedVersion)
|
||||
}
|
||||
}
|
||||
|
||||
etcdThenOps := make([]clientv3.Op, len(onSuccess))
|
||||
for i, o := range onSuccess {
|
||||
switch o.Type {
|
||||
case OpPut:
|
||||
etcdThenOps[i] = clientv3.OpPut(o.Key, string(o.Value))
|
||||
case OpDelete:
|
||||
etcdThenOps[i] = clientv3.OpDelete(o.Key)
|
||||
default:
|
||||
return false, fmt.Errorf("unsupported operation type in transaction 'onSuccess': %v", o.Type)
|
||||
}
|
||||
}
|
||||
|
||||
reqCtx, cancel := context.WithTimeout(ctx, defaultRequestTimeout)
|
||||
defer cancel()
|
||||
|
||||
txn := s.client.Txn(reqCtx)
|
||||
if len(etcdCmps) > 0 {
|
||||
txn = txn.If(etcdCmps...)
|
||||
}
|
||||
txn = txn.Then(etcdThenOps...)
|
||||
// No Else() for general ops, etcd's Else takes clientv3.Op too, but our Op is different.
|
||||
|
||||
resp, err := txn.Commit()
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("etcd transaction commit failed: %w", err)
|
||||
}
|
||||
|
||||
return resp.Succeeded, nil
|
||||
}
|
||||
89
internal/store/interface.go
Normal file
89
internal/store/interface.go
Normal file
@ -0,0 +1,89 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
)
|
||||
|
||||
// KV represents a key-value pair from the store.
|
||||
type KV struct {
|
||||
Key string
|
||||
Value []byte
|
||||
Version int64 // etcd ModRevision or similar versioning
|
||||
}
|
||||
|
||||
// EventType defines the type of change observed by a Watch.
|
||||
type EventType int
|
||||
|
||||
const (
|
||||
// EventTypePut indicates a key was created or updated.
|
||||
EventTypePut EventType = iota
|
||||
// EventTypeDelete indicates a key was deleted.
|
||||
EventTypeDelete
|
||||
)
|
||||
|
||||
// WatchEvent represents a single event from a Watch operation.
|
||||
type WatchEvent struct {
|
||||
Type EventType
|
||||
KV KV
|
||||
PrevKV *KV // Previous KV, if available and applicable (e.g., for updates)
|
||||
}
|
||||
|
||||
// Compare is used in transactions to check a key's version.
|
||||
type Compare struct {
|
||||
Key string
|
||||
ExpectedVersion int64 // 0 means key should not exist. >0 means key must have this version.
|
||||
}
|
||||
|
||||
// OpType defines the type of operation in a transaction.
|
||||
type OpType int
|
||||
|
||||
const (
|
||||
// OpPut represents a put operation.
|
||||
OpPut OpType = iota
|
||||
// OpDelete represents a delete operation.
|
||||
OpDelete
|
||||
// OpGet is not typically used in Txn success/fail ops but included for completeness if needed.
|
||||
OpGet
|
||||
)
|
||||
|
||||
// Op represents an operation to be performed within a transaction.
|
||||
type Op struct {
|
||||
Type OpType
|
||||
Key string
|
||||
Value []byte // Used for OpPut
|
||||
}
|
||||
|
||||
// StateStore defines the interface for interacting with the underlying key-value store.
|
||||
// It's designed based on RFC 5.1.
|
||||
type StateStore interface {
|
||||
// Put stores a key-value pair.
|
||||
Put(ctx context.Context, key string, value []byte) error
|
||||
// Get retrieves a key-value pair. Returns an error if key not found.
|
||||
Get(ctx context.Context, key string) (*KV, error)
|
||||
// Delete removes a key.
|
||||
Delete(ctx context.Context, key string) error
|
||||
// List retrieves all key-value pairs matching a prefix.
|
||||
List(ctx context.Context, prefix string) ([]KV, error)
|
||||
// Watch observes changes to a key or prefix, starting from a given revision.
|
||||
// startRevision = 0 means watch from current.
|
||||
Watch(ctx context.Context, keyOrPrefix string, startRevision int64) (<-chan WatchEvent, error)
|
||||
// Close releases any resources held by the store client.
|
||||
Close() error
|
||||
|
||||
// Campaign attempts to acquire leadership for the given leaderID.
|
||||
// It returns a leadershipCtx that is cancelled when leadership is lost or Resign is called.
|
||||
// leaseTTLSeconds specifies the TTL for the leader's lease.
|
||||
Campaign(ctx context.Context, leaderID string, leaseTTLSeconds int64) (leadershipCtx context.Context, err error)
|
||||
// Resign relinquishes leadership if currently held.
|
||||
// The context passed should ideally be the one associated with the current leadership term or a parent.
|
||||
Resign(ctx context.Context) error
|
||||
// GetLeader retrieves the ID of the current leader.
|
||||
GetLeader(ctx context.Context) (leaderID string, err error)
|
||||
|
||||
// DoTransaction executes a list of operations atomically if all checks pass.
|
||||
// checks are conditions that must be true.
|
||||
// onSuccess operations are performed if checks pass.
|
||||
// onFailure operations are performed if checks fail (not typically supported by etcd Txn else).
|
||||
// Returns true if the transaction was committed (onSuccess ops were applied).
|
||||
DoTransaction(ctx context.Context, checks []Compare, onSuccess []Op, onFailure []Op) (committed bool, err error)
|
||||
}
|
||||
Reference in New Issue
Block a user