diff --git a/controller/db/terminator_store.go b/controller/db/terminator_store.go index 8fee4b214..98b7990b9 100644 --- a/controller/db/terminator_store.go +++ b/controller/db/terminator_store.go @@ -133,13 +133,15 @@ func (entity *Terminator) SetValues(ctx *boltz.PersistContext) { ctx.SetInt32(FieldTerminatorCost, int32(entity.Cost)) ctx.SetRequiredString(FieldTerminatorPrecedence, entity.Precedence) - _ = ctx.Bucket.DeleteBucket([]byte(FieldServerPeerData)) - if entity.PeerData != nil { - hostDataBucket := ctx.Bucket.GetOrCreateBucket(FieldServerPeerData) - for k, v := range entity.PeerData { - key := make([]byte, 4) - binary.LittleEndian.PutUint32(key, k) - hostDataBucket.PutValue(key, v) + if ctx.ProceedWithSet(FieldServerPeerData) { + _ = ctx.Bucket.DeleteBucket([]byte(FieldServerPeerData)) + if entity.PeerData != nil { + hostDataBucket := ctx.Bucket.GetOrCreateBucket(FieldServerPeerData) + for k, v := range entity.PeerData { + key := make([]byte, 4) + binary.LittleEndian.PutUint32(key, k) + hostDataBucket.PutValue(key, v) + } } } diff --git a/controller/db/terminator_store_test.go b/controller/db/terminator_store_test.go index d9ff0c6e6..c34864993 100644 --- a/controller/db/terminator_store_test.go +++ b/controller/db/terminator_store_test.go @@ -20,6 +20,7 @@ import ( "fmt" "github.com/google/uuid" "github.com/openziti/fabric/controller/xt" + "github.com/openziti/foundation/storage/boltz" "github.com/openziti/foundation/util/stringz" "go.etcd.io/bbolt" "math" @@ -38,6 +39,7 @@ func Test_TerminatorStore(t *testing.T) { t.Run("test create/delete terminators", ctx.testLoadQueryTerminators) t.Run("test update terminators", ctx.testUpdateTerminators) t.Run("test delete terminators", ctx.testDeleteTerminators) + t.Run("test patch terminators", ctx.testPatchTerminator) } func (ctx *TestContext) testCreateInvalidTerminators(t *testing.T) { @@ -229,6 +231,7 @@ func (ctx *TestContext) testUpdateTerminators(t *testing.T) { terminator.Service = e.service.Id // service should not be updated ctx.ValidateUpdated(terminator) + } func (ctx *TestContext) testDeleteTerminators(t *testing.T) { @@ -247,6 +250,41 @@ func (ctx *TestContext) testDeleteTerminators(t *testing.T) { ctx.ValidateDeleted(e.terminator.Id) } +func (ctx *TestContext) testPatchTerminator(t *testing.T) { + service := ctx.requireNewService() + router := ctx.requireNewRouter() + + terminator := &Terminator{} + terminator.Service = service.Id + terminator.Router = router.Id + terminator.Binding = uuid.New().String() + terminator.Address = uuid.New().String() + terminator.Cost = 0 + terminator.PeerData = map[uint32][]byte{ + 1: {1, 2, 3}, + 2: {3, 4, 5}, + } + ctx.RequireCreate(terminator) + ctx.ValidateBaseline(terminator) + + terminator.Cost = 100 + checker := boltz.MapFieldChecker{ + FieldTerminatorCost: struct{}{}, + } + ctx.RequirePatch(terminator, checker) + ctx.ValidateUpdated(terminator) + + terminator.PeerData = map[uint32][]byte{ + 1: {7, 8, 9}, + } + + checker = boltz.MapFieldChecker{ + FieldServerPeerData: struct{}{}, + } + ctx.RequirePatch(terminator, checker) + ctx.ValidateUpdated(terminator) +} + type testStrategyFactory struct{} func (t testStrategyFactory) GetStrategyName() string { diff --git a/router/xgress/link_send_buffer.go b/router/xgress/link_send_buffer.go index 07351fbeb..a69df1234 100644 --- a/router/xgress/link_send_buffer.go +++ b/router/xgress/link_send_buffer.go @@ -315,7 +315,7 @@ func (buffer *LinkSendBuffer) retransmit() { } if retransmitted > 0 { - log.Infof("retransmitted [%d] payloads, [%d] buffered, linkSendBufferSize: %d", retransmitted, len(buffer.buffer), buffer.linkSendBufferSize) + log.Debugf("retransmitted [%d] payloads, [%d] buffered, linkSendBufferSize: %d", retransmitted, len(buffer.buffer), buffer.linkSendBufferSize) } buffer.lastRetransmitTime = now } diff --git a/router/xgress/xgress.go b/router/xgress/xgress.go index f4adaf3e7..8d65083d0 100644 --- a/router/xgress/xgress.go +++ b/router/xgress/xgress.go @@ -421,6 +421,11 @@ func (self *Xgress) tx() { func (self *Xgress) flushSendThenClose() { self.CloseTimeout(self.Options.MaxCloseWait) self.ForwardEndOfSession(func(payload *Payload) bool { + if self.payloadBuffer.closed.Get() { + // Avoid spurious 'failed to forward payload' error if the buffer is already closed + return false + } + pfxlog.ContextLogger(self.Label()).Debug("sending end of session payload") return self.forwardPayload(payload) }) @@ -495,7 +500,7 @@ func (self *Xgress) forwardPayload(payload *Payload) bool { sendCallback, err := self.payloadBuffer.BufferPayload(payload) if err != nil { - pfxlog.ContextLogger(self.Label()).WithError(err).Error("failure forwarding payload") + pfxlog.ContextLogger(self.Label()).WithError(err).Error("failure to buffer payload") return false }