2 Unix SMB/CIFS implementation.
4 WINS Replication server
6 Copyright (C) Stefan Metzmacher 2005
8 This program is free software; you can redistribute it and/or modify
9 it under the terms of the GNU General Public License as published by
10 the Free Software Foundation; either version 2 of the License, or
11 (at your option) any later version.
13 This program is distributed in the hope that it will be useful,
14 but WITHOUT ANY WARRANTY; without even the implied warranty of
15 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16 GNU General Public License for more details.
18 You should have received a copy of the GNU General Public License
19 along with this program; if not, write to the Free Software
20 Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
24 #include "dlinklist.h"
25 #include "lib/events/events.h"
26 #include "lib/socket/socket.h"
27 #include "smbd/service_task.h"
28 #include "smbd/service_stream.h"
29 #include "lib/messaging/irpc.h"
30 #include "librpc/gen_ndr/ndr_winsrepl.h"
31 #include "wrepl_server/wrepl_server.h"
32 #include "wrepl_server/wrepl_out_helpers.h"
33 #include "nbt_server/wins/winsdb.h"
34 #include "ldb/include/ldb.h"
35 #include "libcli/composite/composite.h"
36 #include "libcli/wrepl/winsrepl.h"
38 enum wreplsrv_out_connect_stage {
39 WREPLSRV_OUT_CONNECT_STAGE_WAIT_SOCKET,
40 WREPLSRV_OUT_CONNECT_STAGE_WAIT_ASSOC_CTX,
41 WREPLSRV_OUT_CONNECT_STAGE_DONE
44 struct wreplsrv_out_connect_state {
45 enum wreplsrv_out_connect_stage stage;
46 struct composite_context *c;
47 struct wrepl_request *req;
48 struct wrepl_associate assoc_io;
49 enum winsrepl_partner_type type;
50 struct wreplsrv_out_connection *wreplconn;
53 static void wreplsrv_out_connect_handler(struct wrepl_request *req);
55 static NTSTATUS wreplsrv_out_connect_wait_socket(struct wreplsrv_out_connect_state *state)
59 status = wrepl_connect_recv(state->req);
60 NT_STATUS_NOT_OK_RETURN(status);
62 state->req = wrepl_associate_send(state->wreplconn->sock, &state->assoc_io);
63 NT_STATUS_HAVE_NO_MEMORY(state->req);
65 state->req->async.fn = wreplsrv_out_connect_handler;
66 state->req->async.private = state;
68 state->stage = WREPLSRV_OUT_CONNECT_STAGE_WAIT_ASSOC_CTX;
73 static NTSTATUS wreplsrv_out_connect_wait_assoc_ctx(struct wreplsrv_out_connect_state *state)
77 status = wrepl_associate_recv(state->req, &state->assoc_io);
78 NT_STATUS_NOT_OK_RETURN(status);
80 state->wreplconn->assoc_ctx.peer_ctx = state->assoc_io.out.assoc_ctx;
82 if (state->type == WINSREPL_PARTNER_PUSH) {
83 state->wreplconn->partner->push.wreplconn = state->wreplconn;
84 talloc_steal(state->wreplconn->partner, state->wreplconn);
85 } else if (state->type == WINSREPL_PARTNER_PULL) {
86 state->wreplconn->partner->pull.wreplconn = state->wreplconn;
87 talloc_steal(state->wreplconn->partner, state->wreplconn);
90 state->stage = WREPLSRV_OUT_CONNECT_STAGE_DONE;
95 static void wreplsrv_out_connect_handler(struct wrepl_request *req)
97 struct wreplsrv_out_connect_state *state = talloc_get_type(req->async.private,
98 struct wreplsrv_out_connect_state);
99 struct composite_context *c = state->c;
101 switch (state->stage) {
102 case WREPLSRV_OUT_CONNECT_STAGE_WAIT_SOCKET:
103 c->status = wreplsrv_out_connect_wait_socket(state);
105 case WREPLSRV_OUT_CONNECT_STAGE_WAIT_ASSOC_CTX:
106 c->status = wreplsrv_out_connect_wait_assoc_ctx(state);
107 c->state = COMPOSITE_STATE_DONE;
109 case WREPLSRV_OUT_CONNECT_STAGE_DONE:
110 c->status = NT_STATUS_INTERNAL_ERROR;
113 if (!NT_STATUS_IS_OK(c->status)) {
114 c->state = COMPOSITE_STATE_ERROR;
117 if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
122 static struct composite_context *wreplsrv_out_connect_send(struct wreplsrv_partner *partner,
123 enum winsrepl_partner_type type,
124 struct wreplsrv_out_connection *wreplconn)
126 struct composite_context *c = NULL;
127 struct wreplsrv_service *service = partner->service;
128 struct wreplsrv_out_connect_state *state = NULL;
129 struct wreplsrv_out_connection **wreplconnp = &wreplconn;
130 BOOL cached_connection = False;
132 c = talloc_zero(partner, struct composite_context);
135 state = talloc_zero(c, struct wreplsrv_out_connect_state);
136 if (!state) goto failed;
140 c->state = COMPOSITE_STATE_IN_PROGRESS;
141 c->event_ctx = service->task->event_ctx;
142 c->private_data = state;
144 if (type == WINSREPL_PARTNER_PUSH) {
145 cached_connection = True;
146 wreplconn = partner->push.wreplconn;
147 wreplconnp = &partner->push.wreplconn;
148 } else if (type == WINSREPL_PARTNER_PULL) {
149 cached_connection = True;
150 wreplconn = partner->pull.wreplconn;
151 wreplconnp = &partner->pull.wreplconn;
154 /* we have a connection already, so use it */
156 if (!wreplconn->sock->dead) {
157 state->stage = WREPLSRV_OUT_CONNECT_STAGE_DONE;
158 state->wreplconn= wreplconn;
161 } else if (!cached_connection) {
162 state->stage = WREPLSRV_OUT_CONNECT_STAGE_DONE;
163 state->wreplconn= NULL;
167 talloc_free(wreplconn);
172 wreplconn = talloc_zero(state, struct wreplsrv_out_connection);
173 if (!wreplconn) goto failed;
175 wreplconn->service = service;
176 wreplconn->partner = partner;
177 wreplconn->sock = wrepl_socket_init(wreplconn, service->task->event_ctx);
178 if (!wreplconn->sock) goto failed;
180 state->stage = WREPLSRV_OUT_CONNECT_STAGE_WAIT_SOCKET;
181 state->wreplconn= wreplconn;
182 state->req = wrepl_connect_send(wreplconn->sock,
183 partner->our_address,
185 if (!state->req) goto failed;
187 state->req->async.fn = wreplsrv_out_connect_handler;
188 state->req->async.private = state;
196 static NTSTATUS wreplsrv_out_connect_recv(struct composite_context *c, TALLOC_CTX *mem_ctx,
197 struct wreplsrv_out_connection **wreplconn)
201 status = composite_wait(c);
203 if (NT_STATUS_IS_OK(status)) {
204 struct wreplsrv_out_connect_state *state = talloc_get_type(c->private_data,
205 struct wreplsrv_out_connect_state);
206 if (state->wreplconn) {
207 *wreplconn = talloc_reference(mem_ctx, state->wreplconn);
208 if (!*wreplconn) status = NT_STATUS_NO_MEMORY;
210 status = NT_STATUS_INVALID_CONNECTION;
219 enum wreplsrv_pull_table_stage {
220 WREPLSRV_PULL_TABLE_STAGE_WAIT_CONNECTION,
221 WREPLSRV_PULL_TABLE_STAGE_WAIT_TABLE_REPLY,
222 WREPLSRV_PULL_TABLE_STAGE_DONE
225 struct wreplsrv_pull_table_state {
226 enum wreplsrv_pull_table_stage stage;
227 struct composite_context *c;
228 struct wrepl_request *req;
229 struct wrepl_pull_table table_io;
230 struct wreplsrv_pull_table_io *io;
231 struct composite_context *creq;
232 struct wreplsrv_out_connection *wreplconn;
235 static void wreplsrv_pull_table_handler_req(struct wrepl_request *req);
237 static NTSTATUS wreplsrv_pull_table_wait_connection(struct wreplsrv_pull_table_state *state)
241 status = wreplsrv_out_connect_recv(state->creq, state, &state->wreplconn);
242 NT_STATUS_NOT_OK_RETURN(status);
244 state->table_io.in.assoc_ctx = state->wreplconn->assoc_ctx.peer_ctx;
245 state->req = wrepl_pull_table_send(state->wreplconn->sock, &state->table_io);
246 NT_STATUS_HAVE_NO_MEMORY(state->req);
248 state->req->async.fn = wreplsrv_pull_table_handler_req;
249 state->req->async.private = state;
251 state->stage = WREPLSRV_PULL_TABLE_STAGE_WAIT_TABLE_REPLY;
256 static NTSTATUS wreplsrv_pull_table_wait_table_reply(struct wreplsrv_pull_table_state *state)
260 status = wrepl_pull_table_recv(state->req, state, &state->table_io);
261 NT_STATUS_NOT_OK_RETURN(status);
263 state->stage = WREPLSRV_PULL_TABLE_STAGE_DONE;
268 static void wreplsrv_pull_table_handler(struct wreplsrv_pull_table_state *state)
270 struct composite_context *c = state->c;
272 switch (state->stage) {
273 case WREPLSRV_PULL_TABLE_STAGE_WAIT_CONNECTION:
274 c->status = wreplsrv_pull_table_wait_connection(state);
276 case WREPLSRV_PULL_TABLE_STAGE_WAIT_TABLE_REPLY:
277 c->status = wreplsrv_pull_table_wait_table_reply(state);
278 c->state = COMPOSITE_STATE_DONE;
280 case WREPLSRV_PULL_TABLE_STAGE_DONE:
281 c->status = NT_STATUS_INTERNAL_ERROR;
284 if (!NT_STATUS_IS_OK(c->status)) {
285 c->state = COMPOSITE_STATE_ERROR;
288 if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
293 static void wreplsrv_pull_table_handler_creq(struct composite_context *creq)
295 struct wreplsrv_pull_table_state *state = talloc_get_type(creq->async.private_data,
296 struct wreplsrv_pull_table_state);
297 wreplsrv_pull_table_handler(state);
301 static void wreplsrv_pull_table_handler_req(struct wrepl_request *req)
303 struct wreplsrv_pull_table_state *state = talloc_get_type(req->async.private,
304 struct wreplsrv_pull_table_state);
305 wreplsrv_pull_table_handler(state);
309 struct composite_context *wreplsrv_pull_table_send(TALLOC_CTX *mem_ctx, struct wreplsrv_pull_table_io *io)
311 struct composite_context *c = NULL;
312 struct wreplsrv_service *service = io->in.partner->service;
313 struct wreplsrv_pull_table_state *state = NULL;
315 c = talloc_zero(mem_ctx, struct composite_context);
318 state = talloc_zero(c, struct wreplsrv_pull_table_state);
319 if (!state) goto failed;
323 c->state = COMPOSITE_STATE_IN_PROGRESS;
324 c->event_ctx = service->task->event_ctx;
325 c->private_data = state;
327 if (io->in.num_owners) {
328 state->table_io.out.num_partners = io->in.num_owners;
329 state->table_io.out.partners = io->in.owners;
330 state->stage = WREPLSRV_PULL_TABLE_STAGE_DONE;
335 state->stage = WREPLSRV_PULL_TABLE_STAGE_WAIT_CONNECTION;
336 state->creq = wreplsrv_out_connect_send(io->in.partner, WINSREPL_PARTNER_PULL, NULL);
337 if (!state->creq) goto failed;
339 state->creq->async.fn = wreplsrv_pull_table_handler_creq;
340 state->creq->async.private_data = state;
348 NTSTATUS wreplsrv_pull_table_recv(struct composite_context *c, TALLOC_CTX *mem_ctx,
349 struct wreplsrv_pull_table_io *io)
353 status = composite_wait(c);
355 if (NT_STATUS_IS_OK(status)) {
356 struct wreplsrv_pull_table_state *state = talloc_get_type(c->private_data,
357 struct wreplsrv_pull_table_state);
358 io->out.num_owners = state->table_io.out.num_partners;
359 io->out.owners = state->table_io.out.partners;
360 talloc_reference(mem_ctx, state->table_io.out.partners);
367 enum wreplsrv_pull_names_stage {
368 WREPLSRV_PULL_NAMES_STAGE_WAIT_CONNECTION,
369 WREPLSRV_PULL_NAMES_STAGE_WAIT_SEND_REPLY,
370 WREPLSRV_PULL_NAMES_STAGE_DONE
373 struct wreplsrv_pull_names_state {
374 enum wreplsrv_pull_names_stage stage;
375 struct composite_context *c;
376 struct wrepl_request *req;
377 struct wrepl_pull_names pull_io;
378 struct wreplsrv_pull_names_io *io;
379 struct composite_context *creq;
380 struct wreplsrv_out_connection *wreplconn;
383 static void wreplsrv_pull_names_handler_req(struct wrepl_request *req);
385 static NTSTATUS wreplsrv_pull_names_wait_connection(struct wreplsrv_pull_names_state *state)
389 status = wreplsrv_out_connect_recv(state->creq, state, &state->wreplconn);
390 NT_STATUS_NOT_OK_RETURN(status);
392 state->pull_io.in.assoc_ctx = state->wreplconn->assoc_ctx.peer_ctx;
393 state->pull_io.in.partner = state->io->in.owner;
394 state->req = wrepl_pull_names_send(state->wreplconn->sock, &state->pull_io);
395 NT_STATUS_HAVE_NO_MEMORY(state->req);
397 state->req->async.fn = wreplsrv_pull_names_handler_req;
398 state->req->async.private = state;
400 state->stage = WREPLSRV_PULL_NAMES_STAGE_WAIT_SEND_REPLY;
405 static NTSTATUS wreplsrv_pull_names_wait_send_reply(struct wreplsrv_pull_names_state *state)
409 status = wrepl_pull_names_recv(state->req, state, &state->pull_io);
410 NT_STATUS_NOT_OK_RETURN(status);
412 state->stage = WREPLSRV_PULL_NAMES_STAGE_DONE;
417 static void wreplsrv_pull_names_handler(struct wreplsrv_pull_names_state *state)
419 struct composite_context *c = state->c;
421 switch (state->stage) {
422 case WREPLSRV_PULL_NAMES_STAGE_WAIT_CONNECTION:
423 c->status = wreplsrv_pull_names_wait_connection(state);
425 case WREPLSRV_PULL_NAMES_STAGE_WAIT_SEND_REPLY:
426 c->status = wreplsrv_pull_names_wait_send_reply(state);
427 c->state = COMPOSITE_STATE_DONE;
429 case WREPLSRV_PULL_NAMES_STAGE_DONE:
430 c->status = NT_STATUS_INTERNAL_ERROR;
433 if (!NT_STATUS_IS_OK(c->status)) {
434 c->state = COMPOSITE_STATE_ERROR;
437 if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
442 static void wreplsrv_pull_names_handler_creq(struct composite_context *creq)
444 struct wreplsrv_pull_names_state *state = talloc_get_type(creq->async.private_data,
445 struct wreplsrv_pull_names_state);
446 wreplsrv_pull_names_handler(state);
450 static void wreplsrv_pull_names_handler_req(struct wrepl_request *req)
452 struct wreplsrv_pull_names_state *state = talloc_get_type(req->async.private,
453 struct wreplsrv_pull_names_state);
454 wreplsrv_pull_names_handler(state);
458 struct composite_context *wreplsrv_pull_names_send(TALLOC_CTX *mem_ctx, struct wreplsrv_pull_names_io *io)
460 struct composite_context *c = NULL;
461 struct wreplsrv_service *service = io->in.partner->service;
462 struct wreplsrv_pull_names_state *state = NULL;
463 enum winsrepl_partner_type partner_type = WINSREPL_PARTNER_PULL;
465 if (io->in.wreplconn) partner_type = WINSREPL_PARTNER_NONE;
467 c = talloc_zero(mem_ctx, struct composite_context);
470 state = talloc_zero(c, struct wreplsrv_pull_names_state);
471 if (!state) goto failed;
475 c->state = COMPOSITE_STATE_IN_PROGRESS;
476 c->event_ctx = service->task->event_ctx;
477 c->private_data = state;
479 state->stage = WREPLSRV_PULL_NAMES_STAGE_WAIT_CONNECTION;
480 state->creq = wreplsrv_out_connect_send(io->in.partner, partner_type, io->in.wreplconn);
481 if (!state->creq) goto failed;
483 state->creq->async.fn = wreplsrv_pull_names_handler_creq;
484 state->creq->async.private_data = state;
492 NTSTATUS wreplsrv_pull_names_recv(struct composite_context *c, TALLOC_CTX *mem_ctx,
493 struct wreplsrv_pull_names_io *io)
497 status = composite_wait(c);
499 if (NT_STATUS_IS_OK(status)) {
500 struct wreplsrv_pull_names_state *state = talloc_get_type(c->private_data,
501 struct wreplsrv_pull_names_state);
502 io->out.num_names = state->pull_io.out.num_names;
503 io->out.names = state->pull_io.out.names;
504 talloc_reference(mem_ctx, state->pull_io.out.names);
512 enum wreplsrv_pull_cycle_stage {
513 WREPLSRV_PULL_CYCLE_STAGE_WAIT_TABLE_REPLY,
514 WREPLSRV_PULL_CYCLE_STAGE_WAIT_SEND_REPLIES,
515 WREPLSRV_PULL_CYCLE_STAGE_WAIT_STOP_ASSOC,
516 WREPLSRV_PULL_CYCLE_STAGE_DONE
519 struct wreplsrv_pull_cycle_state {
520 enum wreplsrv_pull_cycle_stage stage;
521 struct composite_context *c;
522 struct wreplsrv_pull_cycle_io *io;
523 struct wreplsrv_pull_table_io table_io;
525 struct wreplsrv_pull_names_io names_io;
526 struct composite_context *creq;
527 struct wrepl_associate_stop assoc_stop_io;
528 struct wrepl_request *req;
531 static void wreplsrv_pull_cycle_handler_creq(struct composite_context *creq);
532 static void wreplsrv_pull_cycle_handler_req(struct wrepl_request *req);
534 static NTSTATUS wreplsrv_pull_cycle_next_owner_do_work(struct wreplsrv_pull_cycle_state *state)
536 struct wreplsrv_owner *current_owner;
537 struct wreplsrv_owner *local_owner;
539 uint64_t old_max_version = 0;
540 BOOL do_pull = False;
542 for (i=state->current; i < state->table_io.out.num_owners; i++) {
543 current_owner = wreplsrv_find_owner(state->io->in.partner->pull.table,
544 state->table_io.out.owners[i].address);
546 local_owner = wreplsrv_find_owner(state->io->in.partner->service->table,
547 state->table_io.out.owners[i].address);
549 * this means we are ourself the current owner,
550 * and we don't want replicate ourself
552 if (!current_owner) continue;
555 * this means we don't have any records of this owner
565 * this means the remote partner has some new records of this owner
568 if (current_owner->owner.max_version > local_owner->owner.max_version) {
570 old_max_version = local_owner->owner.max_version;
577 state->names_io.in.partner = state->io->in.partner;
578 state->names_io.in.wreplconn = state->io->in.wreplconn;
579 state->names_io.in.owner = current_owner->owner;
580 state->names_io.in.owner.min_version = old_max_version + 1;
581 state->creq = wreplsrv_pull_names_send(state, &state->names_io);
582 NT_STATUS_HAVE_NO_MEMORY(state->creq);
584 state->creq->async.fn = wreplsrv_pull_cycle_handler_creq;
585 state->creq->async.private_data = state;
587 return STATUS_MORE_ENTRIES;
593 static NTSTATUS wreplsrv_pull_cycle_next_owner_wrapper(struct wreplsrv_pull_cycle_state *state)
597 status = wreplsrv_pull_cycle_next_owner_do_work(state);
598 if (NT_STATUS_IS_OK(status)) {
599 state->stage = WREPLSRV_PULL_CYCLE_STAGE_DONE;
600 } else if (NT_STATUS_EQUAL(STATUS_MORE_ENTRIES, status)) {
601 state->stage = WREPLSRV_PULL_CYCLE_STAGE_WAIT_SEND_REPLIES;
602 status = NT_STATUS_OK;
605 if (state->stage == WREPLSRV_PULL_CYCLE_STAGE_DONE && state->io->in.wreplconn) {
606 state->assoc_stop_io.in.assoc_ctx = state->io->in.wreplconn->assoc_ctx.peer_ctx;
607 state->assoc_stop_io.in.reason = 0;
608 state->req = wrepl_associate_stop_send(state->io->in.wreplconn->sock, &state->assoc_stop_io);
609 NT_STATUS_HAVE_NO_MEMORY(state->req);
611 state->req->async.fn = wreplsrv_pull_cycle_handler_req;
612 state->req->async.private = state;
614 state->stage = WREPLSRV_PULL_CYCLE_STAGE_WAIT_STOP_ASSOC;
620 static NTSTATUS wreplsrv_pull_cycle_wait_table_reply(struct wreplsrv_pull_cycle_state *state)
625 status = wreplsrv_pull_table_recv(state->creq, state, &state->table_io);
626 NT_STATUS_NOT_OK_RETURN(status);
628 /* update partner table */
629 for (i=0; i < state->table_io.out.num_owners; i++) {
632 is_our_addr = wreplsrv_is_our_address(state->io->in.partner->service,
633 state->table_io.out.owners[i].address);
634 if (is_our_addr) continue;
636 status = wreplsrv_add_table(state->io->in.partner->service,
637 state->io->in.partner,
638 &state->io->in.partner->pull.table,
639 state->table_io.out.owners[i].address,
640 state->table_io.out.owners[i].max_version);
641 NT_STATUS_NOT_OK_RETURN(status);
644 status = wreplsrv_pull_cycle_next_owner_wrapper(state);
645 NT_STATUS_NOT_OK_RETURN(status);
650 static NTSTATUS wreplsrv_pull_cycle_apply_records(struct wreplsrv_pull_cycle_state *state)
654 status = wreplsrv_apply_records(state->io->in.partner, &state->names_io);
655 NT_STATUS_NOT_OK_RETURN(status);
657 talloc_free(state->names_io.out.names);
658 ZERO_STRUCT(state->names_io);
663 static NTSTATUS wreplsrv_pull_cycle_wait_send_replies(struct wreplsrv_pull_cycle_state *state)
667 status = wreplsrv_pull_names_recv(state->creq, state, &state->names_io);
668 NT_STATUS_NOT_OK_RETURN(status);
671 * TODO: this should maybe an async call,
672 * because we may need some network access
673 * for conflict resolving
675 status = wreplsrv_pull_cycle_apply_records(state);
676 NT_STATUS_NOT_OK_RETURN(status);
678 status = wreplsrv_pull_cycle_next_owner_wrapper(state);
679 NT_STATUS_NOT_OK_RETURN(status);
684 static NTSTATUS wreplsrv_pull_cycle_wait_stop_assoc(struct wreplsrv_pull_cycle_state *state)
688 status = wrepl_associate_stop_recv(state->req, &state->assoc_stop_io);
689 NT_STATUS_NOT_OK_RETURN(status);
691 state->stage = WREPLSRV_PULL_CYCLE_STAGE_DONE;
696 static void wreplsrv_pull_cycle_handler(struct wreplsrv_pull_cycle_state *state)
698 struct composite_context *c = state->c;
700 switch (state->stage) {
701 case WREPLSRV_PULL_CYCLE_STAGE_WAIT_TABLE_REPLY:
702 c->status = wreplsrv_pull_cycle_wait_table_reply(state);
704 case WREPLSRV_PULL_CYCLE_STAGE_WAIT_SEND_REPLIES:
705 c->status = wreplsrv_pull_cycle_wait_send_replies(state);
707 case WREPLSRV_PULL_CYCLE_STAGE_WAIT_STOP_ASSOC:
708 c->status = wreplsrv_pull_cycle_wait_stop_assoc(state);
710 case WREPLSRV_PULL_CYCLE_STAGE_DONE:
711 c->status = NT_STATUS_INTERNAL_ERROR;
714 if (state->stage == WREPLSRV_PULL_CYCLE_STAGE_DONE) {
715 c->state = COMPOSITE_STATE_DONE;
718 if (!NT_STATUS_IS_OK(c->status)) {
719 c->state = COMPOSITE_STATE_ERROR;
722 if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
727 static void wreplsrv_pull_cycle_handler_creq(struct composite_context *creq)
729 struct wreplsrv_pull_cycle_state *state = talloc_get_type(creq->async.private_data,
730 struct wreplsrv_pull_cycle_state);
731 wreplsrv_pull_cycle_handler(state);
735 static void wreplsrv_pull_cycle_handler_req(struct wrepl_request *req)
737 struct wreplsrv_pull_cycle_state *state = talloc_get_type(req->async.private,
738 struct wreplsrv_pull_cycle_state);
739 wreplsrv_pull_cycle_handler(state);
743 struct composite_context *wreplsrv_pull_cycle_send(TALLOC_CTX *mem_ctx, struct wreplsrv_pull_cycle_io *io)
745 struct composite_context *c = NULL;
746 struct wreplsrv_service *service = io->in.partner->service;
747 struct wreplsrv_pull_cycle_state *state = NULL;
749 c = talloc_zero(mem_ctx, struct composite_context);
752 state = talloc_zero(c, struct wreplsrv_pull_cycle_state);
753 if (!state) goto failed;
757 c->state = COMPOSITE_STATE_IN_PROGRESS;
758 c->event_ctx = service->task->event_ctx;
759 c->private_data = state;
761 state->stage = WREPLSRV_PULL_CYCLE_STAGE_WAIT_TABLE_REPLY;
762 state->table_io.in.partner = io->in.partner;
763 state->table_io.in.num_owners = io->in.num_owners;
764 state->table_io.in.owners = io->in.owners;
765 state->creq = wreplsrv_pull_table_send(state, &state->table_io);
766 if (!state->creq) goto failed;
768 state->creq->async.fn = wreplsrv_pull_cycle_handler_creq;
769 state->creq->async.private_data = state;
777 NTSTATUS wreplsrv_pull_cycle_recv(struct composite_context *c)
781 status = composite_wait(c);
787 enum wreplsrv_push_notify_stage {
788 WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_CONNECT,
789 WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_UPDATE,
790 WREPLSRV_PUSH_NOTIFY_STAGE_DONE
793 struct wreplsrv_push_notify_state {
794 enum wreplsrv_push_notify_stage stage;
795 struct composite_context *c;
796 struct wreplsrv_push_notify_io *io;
797 enum wrepl_replication_cmd command;
799 struct wrepl_request *req;
800 struct wrepl_packet req_packet;
801 struct wrepl_packet *rep_packet;
802 struct composite_context *creq;
803 struct wreplsrv_out_connection *wreplconn;
806 static void wreplsrv_push_notify_handler_creq(struct composite_context *creq);
807 static void wreplsrv_push_notify_handler_req(struct wrepl_request *req);
809 static NTSTATUS wreplsrv_push_notify_update(struct wreplsrv_push_notify_state *state)
811 struct wreplsrv_service *service = state->io->in.partner->service;
812 struct wrepl_packet *req = &state->req_packet;
813 struct wrepl_replication *repl_out = &state->req_packet.message.replication;
814 struct wrepl_table *table_out = &state->req_packet.message.replication.info.table;
815 struct wreplsrv_in_connection *wrepl_in;
817 struct socket_context *sock;
818 struct data_blob_list_item *update_rep;
820 DATA_BLOB update_blob;
822 req->opcode = WREPL_OPCODE_BITS;
823 req->assoc_ctx = state->wreplconn->assoc_ctx.peer_ctx;
824 req->mess_type = WREPL_REPLICATION;
826 repl_out->command = state->command;
828 our_ip = socket_get_my_addr(state->wreplconn->sock->sock, state);
829 NT_STATUS_HAVE_NO_MEMORY(our_ip);
831 status = wreplsrv_fill_wrepl_table(service, state, table_out,
832 our_ip, our_ip, state->full_table);
833 NT_STATUS_NOT_OK_RETURN(status);
835 state->req = wrepl_request_send(state->wreplconn->sock, req);
836 NT_STATUS_HAVE_NO_MEMORY(state->req);
838 sock = state->wreplconn->sock->sock;
839 talloc_steal(state, state->wreplconn->sock->sock);
840 state->wreplconn->sock->sock = NULL;
842 update_blob = state->req->buffer;
843 talloc_steal(state, state->req->buffer.data);
845 talloc_free(state->wreplconn->sock);
846 state->wreplconn->sock = NULL;
848 status = wreplsrv_in_connection_merge(state->io->in.partner,
850 NT_STATUS_NOT_OK_RETURN(status);
852 wrepl_in->assoc_ctx.peer_ctx = state->wreplconn->assoc_ctx.peer_ctx;
853 wrepl_in->assoc_ctx.our_ctx = 0;
855 update_rep = talloc(wrepl_in, struct data_blob_list_item);
856 NT_STATUS_HAVE_NO_MEMORY(update_rep);
858 update_rep->blob = update_blob;
859 talloc_steal(update_rep, update_blob.data);
861 talloc_free(state->wreplconn);
862 state->wreplconn = NULL;
864 if (!wrepl_in->send_queue) {
865 EVENT_FD_WRITEABLE(wrepl_in->conn->event.fde);
867 DLIST_ADD_END(wrepl_in->send_queue, update_rep, struct data_blob_list_item *);
869 state->stage = WREPLSRV_PUSH_NOTIFY_STAGE_DONE;
874 static NTSTATUS wreplsrv_push_notify_inform(struct wreplsrv_push_notify_state *state)
876 struct wreplsrv_service *service = state->io->in.partner->service;
877 struct wrepl_packet *req = &state->req_packet;
878 struct wrepl_replication *repl_out = &state->req_packet.message.replication;
879 struct wrepl_table *table_out = &state->req_packet.message.replication.info.table;
883 req->opcode = WREPL_OPCODE_BITS;
884 req->assoc_ctx = state->wreplconn->assoc_ctx.peer_ctx;
885 req->mess_type = WREPL_REPLICATION;
887 repl_out->command = state->command;
889 our_ip = socket_get_my_addr(state->wreplconn->sock->sock, state);
890 NT_STATUS_HAVE_NO_MEMORY(our_ip);
892 status = wreplsrv_fill_wrepl_table(service, state, table_out,
893 our_ip, our_ip, state->full_table);
894 NT_STATUS_NOT_OK_RETURN(status);
896 state->req = wrepl_request_send(state->wreplconn->sock, req);
897 NT_STATUS_HAVE_NO_MEMORY(state->req);
899 /* we won't get a reply to a inform message */
900 state->req->send_only = True;
901 state->req->async.fn = wreplsrv_push_notify_handler_req;
902 state->req->async.private = state;
904 state->stage = WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_UPDATE;
909 static NTSTATUS wreplsrv_push_notify_wait_connect(struct wreplsrv_push_notify_state *state)
913 status = wreplsrv_out_connect_recv(state->creq, state, &state->wreplconn);
914 NT_STATUS_NOT_OK_RETURN(status);
916 switch (state->command) {
917 case WREPL_REPL_UPDATE:
918 state->full_table = True;
919 return wreplsrv_push_notify_update(state);
920 case WREPL_REPL_UPDATE2:
921 state->full_table = False;
922 return wreplsrv_push_notify_update(state);
923 case WREPL_REPL_INFORM:
924 state->full_table = True;
925 return wreplsrv_push_notify_inform(state);
926 case WREPL_REPL_INFORM2:
927 state->full_table = False;
928 return wreplsrv_push_notify_inform(state);
930 return NT_STATUS_INTERNAL_ERROR;
933 return NT_STATUS_INTERNAL_ERROR;
936 static NTSTATUS wreplsrv_push_notify_wait_update(struct wreplsrv_push_notify_state *state)
938 return NT_STATUS_FOOBAR;
941 static void wreplsrv_push_notify_handler(struct wreplsrv_push_notify_state *state)
943 struct composite_context *c = state->c;
945 switch (state->stage) {
946 case WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_CONNECT:
947 c->status = wreplsrv_push_notify_wait_connect(state);
949 case WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_UPDATE:
950 c->status = wreplsrv_push_notify_wait_update(state);
952 case WREPLSRV_PUSH_NOTIFY_STAGE_DONE:
953 c->status = NT_STATUS_INTERNAL_ERROR;
956 if (state->stage == WREPLSRV_PUSH_NOTIFY_STAGE_DONE) {
957 c->state = COMPOSITE_STATE_DONE;
960 if (!NT_STATUS_IS_OK(c->status)) {
961 c->state = COMPOSITE_STATE_ERROR;
964 if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
969 static void wreplsrv_push_notify_handler_creq(struct composite_context *creq)
971 struct wreplsrv_push_notify_state *state = talloc_get_type(creq->async.private_data,
972 struct wreplsrv_push_notify_state);
973 wreplsrv_push_notify_handler(state);
977 static void wreplsrv_push_notify_handler_req(struct wrepl_request *req)
979 struct wreplsrv_push_notify_state *state = talloc_get_type(req->async.private,
980 struct wreplsrv_push_notify_state);
981 wreplsrv_push_notify_handler(state);
985 struct composite_context *wreplsrv_push_notify_send(TALLOC_CTX *mem_ctx, struct wreplsrv_push_notify_io *io)
987 struct composite_context *c = NULL;
988 struct wreplsrv_service *service = io->in.partner->service;
989 struct wreplsrv_push_notify_state *state = NULL;
990 enum winsrepl_partner_type partner_type;
992 c = talloc_zero(mem_ctx, struct composite_context);
995 state = talloc_zero(c, struct wreplsrv_push_notify_state);
996 if (!state) goto failed;
1000 if (io->in.inform) {
1001 /* we can cache the connection in partner->push->wreplconn */
1002 partner_type = WINSREPL_PARTNER_PUSH;
1003 if (io->in.propagate) {
1004 state->command = WREPL_REPL_INFORM2;
1006 state->command = WREPL_REPL_INFORM;
1009 /* we can NOT cache the connection */
1010 partner_type = WINSREPL_PARTNER_NONE;
1011 if (io->in.propagate) {
1012 state->command = WREPL_REPL_UPDATE2;
1014 state->command = WREPL_REPL_UPDATE;
1018 c->state = COMPOSITE_STATE_IN_PROGRESS;
1019 c->event_ctx = service->task->event_ctx;
1020 c->private_data = state;
1022 state->stage = WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_CONNECT;
1023 state->creq = wreplsrv_out_connect_send(io->in.partner, partner_type, NULL);
1024 if (!state->creq) goto failed;
1026 state->creq->async.fn = wreplsrv_push_notify_handler_creq;
1027 state->creq->async.private_data = state;
1035 NTSTATUS wreplsrv_push_notify_recv(struct composite_context *c)
1039 status = composite_wait(c);