fixed delayuntil in kauri's begin

go.mod

@@ -1,26 +1,26 @@
 module github.com/relab/hotstuff
 
-go 1.25.4
+go 1.25.6
 
 require (
-	cuelang.org/go v0.15.0
+	cuelang.org/go v0.15.4
 	github.com/felixge/fgprof v0.9.5
 	github.com/google/go-cmp v0.7.0
 	github.com/kilic/bls12-381 v0.1.1-0.20210208205449-6045b0235e36
 	github.com/mroth/weightedrand v1.0.0
 	github.com/relab/gorums v0.10.0
 	github.com/relab/iago v0.0.0-20251028232537-e5b08eb0c08b
 	github.com/relab/wrfs v0.0.0-20220416082020-a641cd350078
-	github.com/spf13/cobra v1.10.1
+	github.com/spf13/cobra v1.10.2
 	github.com/spf13/viper v1.21.0
 	go-hep.org/x/hep v0.38.1
-	go.uber.org/zap v1.27.0
-	golang.org/x/term v0.37.0
+	go.uber.org/zap v1.27.1
+	golang.org/x/term v0.39.0
 	golang.org/x/time v0.14.0
 	gonum.org/v1/plot v0.16.0
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20251111163417-95abcf5c77ba
-	google.golang.org/grpc v1.76.0
-	google.golang.org/protobuf v1.36.10
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20260126211449-d11affda4bed
+	google.golang.org/grpc v1.78.0
+	google.golang.org/protobuf v1.36.11
 )
 
 require (
@@ -34,9 +34,9 @@ require (
 	github.com/cockroachdb/apd/v3 v3.2.1 // indirect
 	github.com/emicklei/proto v1.14.2 // indirect
 	github.com/fsnotify/fsnotify v1.9.0 // indirect
-	github.com/go-viper/mapstructure/v2 v2.4.0 // indirect
+	github.com/go-viper/mapstructure/v2 v2.5.0 // indirect
 	github.com/golang/freetype v0.0.0-20170609003504-e2365dfdc4a0 // indirect
-	github.com/google/pprof v0.0.0-20251007162407-5df77e3f7d1d // indirect
+	github.com/google/pprof v0.0.0-20260115054156-294ebfa9ad83 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/kevinburke/ssh_config v1.4.0 // indirect
@@ -47,8 +47,8 @@ require (
 	github.com/pelletier/go-toml/v2 v2.2.4 // indirect
 	github.com/pkg/sftp v1.13.10 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
-	github.com/protocolbuffers/txtpbfmt v0.0.0-20251016062345-16587c79cd91 // indirect
-	github.com/relab/container v0.0.0-20251028224705-baa7b7c5c895 // indirect
+	github.com/protocolbuffers/txtpbfmt v0.0.0-20251124094003-fcb97cc64c7b // indirect
+	github.com/relab/container v0.0.0-20260109140004-4adfae874bb5 // indirect
 	github.com/rogpeppe/go-internal v1.14.1 // indirect
 	github.com/sagikazarmark/locafero v0.12.0 // indirect
 	github.com/spf13/afero v1.15.0 // indirect
@@ -58,16 +58,16 @@ require (
 	github.com/tetratelabs/wazero v1.10.1 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
-	golang.org/x/crypto v0.44.0 // indirect
-	golang.org/x/image v0.33.0 // indirect
-	golang.org/x/mod v0.30.0 // indirect
-	golang.org/x/net v0.47.0 // indirect
+	golang.org/x/crypto v0.47.0 // indirect
+	golang.org/x/image v0.35.0 // indirect
+	golang.org/x/mod v0.31.0 // indirect
+	golang.org/x/net v0.49.0 // indirect
 	golang.org/x/oauth2 v0.33.0 // indirect
-	golang.org/x/sync v0.18.0 // indirect
-	golang.org/x/sys v0.38.0 // indirect
-	golang.org/x/text v0.31.0 // indirect
-	golang.org/x/tools v0.39.0 // indirect
-	gonum.org/v1/gonum v0.16.0 // indirect
+	golang.org/x/sync v0.19.0 // indirect
+	golang.org/x/sys v0.40.0 // indirect
+	golang.org/x/text v0.33.0 // indirect
+	golang.org/x/tools v0.40.0 // indirect
+	gonum.org/v1/gonum v0.17.0 // indirect
 	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

internal/testutil/mocksender.go

@@ -10,11 +10,18 @@ import (
 	"github.com/relab/hotstuff/security/blockchain"
 )
 
+// ContributionMsg represents a contribution sent to parent in Kauri protocol.
+type ContributionMsg struct {
+	View hotstuff.View
+	QC   hotstuff.QuorumSignature
+}
+
 type MockSender struct {
-	id           hotstuff.ID
-	recipients   []hotstuff.ID
-	messagesSent []any
-	blockChains  []*blockchain.Blockchain
+	id            hotstuff.ID
+	recipients    []hotstuff.ID
+	messagesSent  []any
+	contributions []ContributionMsg
+	blockChains   []*blockchain.Blockchain
 }
 
 // NewMockSender returns a mock implementation of core.Sender that
@@ -111,7 +118,18 @@ func (m *MockSender) Sub(ids []hotstuff.ID) (core.Sender, error) {
 	}, nil
 }
 
+// SendContributionToParent stores a contribution message for Kauri protocol testing.
+func (m *MockSender) SendContributionToParent(view hotstuff.View, qc hotstuff.QuorumSignature) {
+	m.contributions = append(m.contributions, ContributionMsg{View: view, QC: qc})
+}
+
+// ContributionsSent returns a slice of contributions sent by the mock sender.
+func (m *MockSender) ContributionsSent() []ContributionMsg {
+	return m.contributions
+}
+
 var _ core.Sender = (*MockSender)(nil)
+var _ core.KauriSender = (*MockSender)(nil)
 
 func isSubset(a, b []hotstuff.ID) bool {
 	set := make(map[hotstuff.ID]struct{}, len(a))

protocol/comm/kauri.go

@@ -9,6 +9,7 @@ import (
 	"github.com/relab/hotstuff/core/eventloop"
 	"github.com/relab/hotstuff/core/logging"
 	"github.com/relab/hotstuff/internal/proto/hotstuffpb"
+	"github.com/relab/hotstuff/internal/proto/kauripb"
 	"github.com/relab/hotstuff/internal/tree"
 	"github.com/relab/hotstuff/network"
 	"github.com/relab/hotstuff/protocol/comm/kauri"
@@ -61,12 +62,15 @@ func NewKauri(
 	eventloop.Register(el, func(_ hotstuff.ReplicaConnectedEvent) {
 		k.initDone = true // signal that we are connected
 	})
-	eventloop.Register(el, func(event kauri.ContributionRecvEvent) {
+	eventloop.Register(el, func(event *kauripb.Contribution) {
 		k.onContributionRecv(event)
 	})
 	eventloop.Register(el, func(event WaitTimerExpiredEvent) {
 		k.onWaitTimerExpired(event)
 	})
+	eventloop.Register(el, func(event WaitForConnectedEvent) {
+		k.onWaitForConnected(event)
+	})
 	return k
 }
 
@@ -83,11 +87,9 @@ func (k *Kauri) Aggregate(proposal *hotstuff.ProposeMsg, pc hotstuff.PartialCert
 // begin starts dissemination of proposal and aggregation of votes.
 func (k *Kauri) begin(p *hotstuff.ProposeMsg, pc hotstuff.PartialCert) error {
 	if !k.initDone {
-		// TODO(meling): This is not correct use of DelayUntil, see issue #267
-		eventloop.DelayUntil[network.ConnectedEvent](k.eventLoop, func() {
-			if err := k.begin(p, pc); err != nil {
-				k.logger.Error(err)
-			}
+		eventloop.DelayUntil[network.ConnectedEvent](k.eventLoop, WaitForConnectedEvent{
+			pc: pc,
+			p:  p,
 		})
 		return nil
 	}
@@ -128,12 +130,24 @@ func (k *Kauri) waitToAggregate() {
 	k.eventLoop.AddEvent(WaitTimerExpiredEvent{currentView: view})
 }
 
+// onWaitForConnected is invoked when begin is called before the replica is connected.
+func (k *Kauri) onWaitForConnected(event WaitForConnectedEvent) {
+	k.logger.Debugf("WaitForConnectedEvent: %v", event)
+	if k.currentView > hotstuff.View(event.p.Block.View()) {
+		k.logger.Debug("Current view is higher than event view, not starting kauri")
+		return
+	}
+	err := k.begin(event.p, event.pc)
+	if err != nil {
+		k.logger.Errorf("Failed to begin kauri after connection: %v", err)
+	}
+}
+
 // onContributionRecv is invoked upon receiving the vote for aggregation.
-func (k *Kauri) onContributionRecv(event kauri.ContributionRecvEvent) {
-	if k.currentView != hotstuff.View(event.Contribution.View) {
+func (k *Kauri) onContributionRecv(contribution *kauripb.Contribution) {
+	if k.currentView != hotstuff.View(contribution.View) {
 		return
 	}
-	contribution := event.Contribution
 	k.logger.Debugf("Processing the contribution from %d", contribution.ID)
 	currentSignature := hotstuffpb.QuorumSignatureFromProto(contribution.Signature)
 	err := k.mergeContribution(currentSignature)
@@ -196,4 +210,9 @@ type WaitTimerExpiredEvent struct {
 	currentView hotstuff.View
 }
 
+type WaitForConnectedEvent struct {
+	pc hotstuff.PartialCert
+	p  *hotstuff.ProposeMsg
+}
+
 var _ Communication = (*Kauri)(nil)

protocol/comm/kauri/service.go

@@ -20,10 +20,5 @@ func RegisterService(
 }
 
 func (i kauriServiceImpl) SendContribution(_ gorums.ServerCtx, request *kauripb.Contribution) {
-	i.eventLoop.AddEvent(ContributionRecvEvent{Contribution: request})
-}
-
-// ContributionRecvEvent is raised when a contribution is received.
-type ContributionRecvEvent struct {
-	Contribution *kauripb.Contribution
+	i.eventLoop.AddEvent(request)
 }