added visibility_timeout.go #11
Conversation
…nd time extension logic
There was a problem hiding this comment.
Commented. Please move this handlers to the server.go and maybe you can focus more on how to implement the visibility timeout logic so that all nodes can track whether that message is safe to stream to the client without depending on spanner data, maybe through broadcast?
| } | ||
|
|
||
| // Publish writes a message to Spanner and acknowledges receipt | ||
| func (s *PubSubServer) Publish(ctx context.Context, req *pubsubpb.PublishRequest) (*pubsubpb.PublishResponse, error) { |
There was a problem hiding this comment.
I think we have already Publish API?
There was a problem hiding this comment.
Please edit the current Publish that we have if you need to do some changes.
| return &pubsubpb.ModifyVisibilityTimeoutResponse{Success: true}, nil | ||
| } | ||
|
|
||
| func main() { |
There was a problem hiding this comment.
Could be our main func in main.go
| func (s *PubSubServer) Subscribe(req *pubsubpb.SubscribeRequest, stream pubsubpb.PubSubService_SubscribeServer) error { | ||
| ctx := stream.Context() | ||
| stmt := spanner.Statement{ | ||
| SQL: `SELECT id, payload, topic, visibilityTimeout, processed FROM Messages |
There was a problem hiding this comment.
Should not query directly to spanner, instead utilize the data received from the leader.
| return err | ||
| } | ||
|
|
||
| msg.Processed = false |
There was a problem hiding this comment.
No need to set to false since all messages from the leader has processed = false.
|
|
||
| // Acknowledge confirms message processing | ||
| func (s *PubSubServer) Acknowledge(ctx context.Context, req *pubsubpb.AcknowledgeRequest) (*pubsubpb.AcknowledgeResponse, error) { | ||
| _, err := s.spannerClient.Apply(ctx, []*spanner.Mutation{ |
There was a problem hiding this comment.
After receiving Ack request, it's supposed to remove the messages from the queue right? then set the status in spanner.
|
good day sir, i have updated the server.go in my branch that primarily focuses on implementing the visibility timeout logic while ensuring that all nodes track whether a message is safe to stream to a client without solely relying on Spanner. It maintains an in-memory structure to store message visibility states, allowing nodes to coordinate through broadcast mechanisms rather than frequent database queries. When a subscriber requests a message, the system first checks if the message is locked or visible. If the message is available, it is assigned exclusively to the subscriber, and a visibility timeout is set. This timeout is synchronized across all nodes using a broadcast mechanism to ensure consistency. If the subscriber acknowledges the message, it is removed from the in-memory queue and marked as processed in Spanner. If the subscriber fails to acknowledge before the timeout expires, the message is made available again for other subscribers. Additionally, the system allows subscribers to request visibility extensions, updating both the in-memory state and Spanner asynchronously. The subscription process does not query Spanner directly but instead retrieves messages from the leader node’s in-memory queue, ensuring efficient message distribution. This approach enforces exclusive message locks per subscriber while minimizing database dependency and reducing query load. are my codes on the right track sir? |
server.go
Outdated
| messageQueueMu sync.RWMutex | ||
| } | ||
|
|
||
| type broadCastInput struct { |
There was a problem hiding this comment.
Is it possible to use the one in broadcast.go?
server.go
Outdated
| visibilityTimeouts sync.Map // messageID -> VisibilityInfo | ||
| lockMu sync.RWMutex | ||
|
|
||
| messageQueue map[string][]*pb.Message // topic -> messages | ||
| messageQueueMu sync.RWMutex |
There was a problem hiding this comment.
Please move this to PubSub object.
There was a problem hiding this comment.
move it from the server struct into a separate PubSub struct. is this what you mean sir?
maybe like this?
type PubSub struct {
visibilityTimeouts sync.Map // messageID -> VisibilityInfo
lockMu sync.RWMutex
messageQueue map[string][]*pb.Message // topic -> messages
messageQueueMu sync.RWMutex
}
type server struct {
client *spanner.Client
op *hedge.Op
pb.UnimplementedPubSubServiceServer
pubSub *PubSub
}
server.go
Outdated
| MessageID string `json:"messageId"` | ||
| SubscriberID string `json:"subscriberId"` | ||
| ExpiresAt time.Time `json:"expiresAt"` | ||
| NodeID string `json:"nodeId"` |
There was a problem hiding this comment.
May I now what is NodeID for? It seems it's an uuid and generated every locking of message. So I assume it's different per message?
There was a problem hiding this comment.
NodeID is generated as a UUID when locking a message. it is different per message, it means each message lock instance gets a unique identifier.
There was a problem hiding this comment.
I see, when can nodeID be used?
There was a problem hiding this comment.
it can be useful in distributed environments where multiple nodes manage message processing.
If multiple nodes handle subscriptions, NodeID can help track which node locked a message. Example: If NodeID is stored with VisibilityInfo, a node can check if it owns a lock before attempting an update.
Before allowing a visibility timeout extension, we could verify if the request comes from the correct node using NodeID. Example: Only the original locking node (with the correct NodeID) can extend the timeout.
If issues arise where messages are incorrectly locked/unlocked, NodeID can help trace which node last held the message.
There was a problem hiding this comment.
should i remove it sir?
There was a problem hiding this comment.
let's discuss this further.
server.go
Outdated
| ) | ||
|
|
||
| func NewServer(client *spanner.Client, op *hedge.Op) *server { |
There was a problem hiding this comment.
No need to initialize server
There was a problem hiding this comment.
You can try to do it in main.go
server.go
Outdated
| "topic": req.Topic, | ||
| }, | ||
| } | ||
| if err := s.broadcastAck(bcastin); err != nil { |
There was a problem hiding this comment.
It seems that there are no definitions for broadcastAck()
broadcast.go
Outdated
| parts := strings.Split(string(msg), ":") | ||
| switch parts[0] { | ||
| case "lock": | ||
| //if a node receives a "lock" request for a message it already has locked, it should reject duplicate locks. | ||
| messageID := parts[1] | ||
| if _, exists := app.messageLocks.Load(messageID); exists { | ||
| return nil, nil // Already locked, ignore duplicate | ||
| } | ||
| case "unlock": | ||
| // Handle unlock request | ||
| messageID := parts[1] | ||
| app.messageLocks.Delete(messageID) | ||
| // Clean up locks and timers | ||
| case "delete": | ||
| messageID := parts[1] | ||
| app.messageLocks.Delete(messageID) | ||
| app.messageQueue.Delete(messageID) | ||
| case "extend": | ||
| // Handle timeout extension | ||
| messageID := parts[1] | ||
| newTimeout, _ := strconv.Atoi(parts[2]) | ||
| if lockInfo, ok := app.messageLocks.Load(messageID); ok { | ||
| info := lockInfo.(MessageLockInfo) | ||
| info.Timeout = time.Now().Add(time.Duration(newTimeout) * time.Second) | ||
| app.messageLocks.Store(messageID, info) | ||
| } | ||
| // Update timeout and reset timer |
There was a problem hiding this comment.
@kitkatchoco2002 Please create function per case.
There was a problem hiding this comment.
the handleBroadcastedMsg is supposedly for type = message only, you can add you own type and create separate function each.
app.go
Outdated
| messageLocks sync.Map // messageID -> MessageLockInfo | ||
| messageQueue sync.Map // topic -> []*pb.Message | ||
| subscriptions sync.Map // subscriptionID -> *pb.Subscription | ||
|
|
||
| // Timer tracking | ||
| timeoutTimers sync.Map // messageID -> *time.Timer |
There was a problem hiding this comment.
Please try to coordinate with this with Jansen, since he will be responsible for storing in-mem data.
helper functions for the server.go
…he leader to all nodes
…ndleBroadcastedMsg) and handleMessageTimeout (helpers.go)
Horichii
force-pushed
the
kate_branch
branch
from
1f416a5 to
2e8255e
Compare
Implement message distribution, acknowledgment, visibility timeout, and time extension logic