Skip to content

initial commit for new client #294

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

Draft
hanish520 wants to merge 5 commits into
base: main
Choose a base branch
from
Draft

initial commit for new client #294

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
a295874
initial commit for new client
hanish520 Jan 12, 2026
90c6237
changed the commandstatus to quorum call
hanish520 Jan 13, 2026
659872d
added tests for command status tracker
hanish520 Jan 16, 2026
3482caa
chore: go mod tidy and regenerate proto files
meling Jan 28, 2026
38035b8
Modified the client design based on the discussion(still WIP)
hanish520 Feb 2, 2026
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
99 changes: 81 additions & 18 deletions client/client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import (
"errors"
"io"
"math"
"strings"
"sync"
"time"

Expand All @@ -21,7 +22,6 @@ import (
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/protobuf/types/known/emptypb"
)

// ID is the identifier for a client.
Expand All @@ -31,17 +31,37 @@ type qspec struct {
faulty int
}

func (q *qspec) ExecCommandQF(_ *clientpb.Command, signatures map[uint32]*emptypb.Empty) (*emptypb.Empty, bool) {
if len(signatures) < q.faulty+1 {
func (q *qspec) CommandStatusQF(command *clientpb.Command, replies map[uint32]*clientpb.CommandStatusResponse) (*clientpb.CommandStatusResponse, bool) {

@meling meling Jan 28, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

See also my comment below; I think this should be redesigned to keep track of how many commands each replica has executed and combine the results into a single reply that tells the client how many of its commands have actually been executed. For example, something along these lines (I'm not checking for nil or other failure modes here):

lowMin, highMin := uint64(0), uint64(math.MaxUint64)
for _, resp := range replies {
	if resp.LowSequenceNumber != command.LowSequenceNumber {
		// invalid reply??
	}
	lowMin = min(lowMin, resp.LowSequenceNumber)
	highMin = min(highMin, resp.HighSequenceNumber)
}

The logic above isn't quite right, since we need f+1 "proofs" of the highMin value.

if len(replies) < q.faulty+1 {
return nil, false
}
return &emptypb.Empty{}, true
responseCount := make([]int, 4) // assuming 4 possible statuses

for _, resp := range replies {
if resp != nil {
status := resp.Status
if int(status) >= 0 && int(status) < len(responseCount) {
responseCount[int(status)]++
}
}
}
for status, count := range responseCount {
if count >= q.faulty+1 {
return &clientpb.CommandStatusResponse{
Command: command,
Status: clientpb.CommandStatusResponse_Status(status),
}, true
}
}
return &clientpb.CommandStatusResponse{
Command: command,
Status: clientpb.CommandStatusResponse_PENDING,
}, false
}

type pendingCmd struct {
sequenceNumber uint64
sendTime time.Time
promise *clientpb.AsyncEmpty
cancelCtx context.CancelFunc
}

Expand Down Expand Up @@ -125,6 +145,7 @@ func (c *Client) Connect(replicas []hotstuff.ReplicaInfo) (err error) {
}
c.gorumsConfig, err = c.mgr.NewConfiguration(&qspec{faulty: hotstuff.NumFaulty(len(replicas))}, gorums.WithNodeMap(nodes))
if err != nil {
c.logger.Error("unable to create the configuration in client")
c.mgr.Close()
return err
}
Expand Down Expand Up @@ -182,6 +203,8 @@ func (c *Client) Stop() {
}

func (c *Client) close() {
// Signal the command handler to stop fetching statuses before closing the manager.
close(c.pendingCmds)
c.mgr.Close()
err := c.reader.Close()
if err != nil {
Expand Down Expand Up @@ -238,8 +261,8 @@ loop:
}

ctx, cancel := context.WithTimeout(ctx, c.timeout)
promise := c.gorumsConfig.ExecCommand(ctx, cmd)
pending := pendingCmd{sequenceNumber: num, sendTime: time.Now(), promise: promise, cancelCtx: cancel}
c.gorumsConfig.ExecCommand(ctx, cmd)
pending := pendingCmd{sequenceNumber: num, sendTime: time.Now(), cancelCtx: cancel}

num++
select {
Expand All @@ -257,6 +280,46 @@ loop:
return nil
}

func (c *Client) fetchCommandStatus(sequenceNumber uint64) hotstuff.CommandStatus {
ticker := time.NewTicker(100 * time.Millisecond)

@meling meling Jan 28, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

This code repeats the CommandStatus call several times, losing the benefit of separating ExecCommand (multicast dissemination) from the CommandStatus call (result checking). Removing the loop and the ticker is the right approach here. Something like this worked for me:

func (c *Client) fetchCommandStatus(sequenceNumber uint64) hotstuff.CommandStatus {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	// ctx, cancel := context.WithTimeout(context.Background(), c.timeout)
	defer cancel()

	cmd := &clientpb.Command{
		ClientID:       uint32(c.id),
		SequenceNumber: sequenceNumber,
	}
	response, err := c.gorumsConfig.CommandStatus(ctx, cmd)
	if err != nil {
		c.logger.Errorf("Failed to fetch command status (client: %d, sequence: %d): %v", c.id, sequenceNumber, err)
		return hotstuff.UNKNOWN
	}
	if response == nil || response.Command == nil {
		c.logger.Errorf("Invalid response received when fetching command status (client: %d, sequence: %d)", c.id, sequenceNumber)
		return hotstuff.UNKNOWN
	}
	c.logger.Infof("Fetched command status (client: %d, sequence: %d, status: %d)", c.id, sequenceNumber, response.Status)
	status := hotstuff.CommandStatus(response.Status)
	if status == hotstuff.COMMITTED || status == hotstuff.EXECUTED || status == hotstuff.FAILED {
		return status
	}
	return hotstuff.UNKNOWN
}

That is, we use a timeout context to cancel the call if the quorum call didn't finish in time. Note that I tried using c.timeout instead of 5 seconds, but the default c.timeout is only 500 ms, which was too short for this call.

Remember that the quorum call is also processed in the CommandStatusQF, and we should normally receive a quorum of replies without having to repeat the call multiple times. If we don't get a reply, it's likely the replicas are overloaded, and repeating the call here won't help.

@meling meling Jan 28, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

You probably want to handle errors something like this:

if errors.Is(err, context.DeadlineExceeded) {
	c.logger.Debug("Command timed out.")
	timeout++
} else if !errors.Is(err, context.Canceled) {
	c.logger.Debugf("Did not get enough replies for command: %v\n", err)
	failed++
}

defer ticker.Stop()
timeout := time.After(c.timeout)

for {
select {
case <-ticker.C:
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
cmd := &clientpb.Command{
ClientID: uint32(c.id),
SequenceNumber: sequenceNumber,
}

response, err := c.gorumsConfig.CommandStatus(ctx, cmd)
cancel()

if err != nil {
c.logger.Errorf("Failed to fetch command status (client: %d, sequence: %d): %v", c.id, sequenceNumber, err)
// If the node/manager was closed, stop trying and return UNKNOWN.
if strings.Contains(err.Error(), "node closed") {
return hotstuff.UNKNOWN
}
continue
}
if response == nil || response.Command == nil {
c.logger.Errorf("Invalid response received when fetching command status (client: %d, sequence: %d)", c.id, sequenceNumber)
continue
}
c.logger.Infof("Fetched command status (client: %d, sequence: %d, status: %d)", c.id, sequenceNumber, response.Status)
status := hotstuff.CommandStatus(response.Status)
if status == hotstuff.COMMITTED || status == hotstuff.EXECUTED || status == hotstuff.FAILED {
return status
}
case <-timeout:
return hotstuff.UNKNOWN
}
}
}

// handleCommands will get pending commands from the pendingCmds channel and then
// handle them as they become acknowledged by the replicas. We expect the commands to be
// acknowledged in the order that they were sent.
Expand All @@ -274,24 +337,24 @@ func (c *Client) handleCommands(ctx context.Context) (executed, failed, timeout
case <-ctx.Done():
return
}
_, err := cmd.promise.Get()
if err != nil {
if errors.Is(err, context.DeadlineExceeded) {
c.logger.Debug("Command timed out.")
timeout++
} else if !errors.Is(err, context.Canceled) {
c.logger.Debugf("Did not get enough replies for command: %v\n", err)
failed++
}
} else {
response := c.fetchCommandStatus(cmd.sequenceNumber)

@meling meling Jan 28, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Another note about performance. We are now actually calling ExecCommand followed by a CommandStatus quorum call, essentially doubling the "effort" that the replicas have to deal with.

I thought the idea was to use ExecCommand for many client commands in a loop, and then periodically (e.g., every 5 seconds) call CommandStatus to ask the replicas to provide "proof of execution" for a sequence of the client's commands. (We can simplify the "proof of execution" to just send back the range for the ClientID that the replicas have executed... and later we can add proper proofs.)


switch response {
case hotstuff.UNKNOWN:
c.logger.Infof("Command timed out (client: %d, sequence: %d)", c.id, cmd.sequenceNumber)
timeout++
case hotstuff.FAILED:
c.logger.Infof("Command failed (client: %d, sequence: %d)", c.id, cmd.sequenceNumber)
failed++
default:
c.logger.Infof("Command executed (client: %d, sequence: %d)", c.id, cmd.sequenceNumber)
executed++
}
c.mut.Lock()
if cmd.sequenceNumber > c.highestCommitted {
c.highestCommitted = cmd.sequenceNumber
}
c.mut.Unlock()

duration := time.Since(cmd.sendTime)
c.eventLoop.AddEvent(LatencyMeasurementEvent{Latency: duration})
}
Expand Down
Loading