Merge branch 'v4-0-test' of ssh://git.samba.org/data/git/samba into registry
[kai/samba.git] / source4 / wrepl_server / wrepl_out_helpers.c
1 /* 
2    Unix SMB/CIFS implementation.
3    
4    WINS Replication server
5    
6    Copyright (C) Stefan Metzmacher      2005
7    
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 3 of the License, or
11    (at your option) any later version.
12    
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.
17    
18    You should have received a copy of the GNU General Public License
19    along with this program.  If not, see <http://www.gnu.org/licenses/>.
20 */
21
22 #include "includes.h"
23 #include "lib/events/events.h"
24 #include "lib/socket/socket.h"
25 #include "smbd/service_task.h"
26 #include "smbd/service_stream.h"
27 #include "librpc/gen_ndr/winsrepl.h"
28 #include "wrepl_server/wrepl_server.h"
29 #include "nbt_server/wins/winsdb.h"
30 #include "libcli/composite/composite.h"
31 #include "libcli/wrepl/winsrepl.h"
32 #include "libcli/resolve/resolve.h"
33 #include "param/param.h"
34
35 enum wreplsrv_out_connect_stage {
36         WREPLSRV_OUT_CONNECT_STAGE_WAIT_SOCKET,
37         WREPLSRV_OUT_CONNECT_STAGE_WAIT_ASSOC_CTX,
38         WREPLSRV_OUT_CONNECT_STAGE_DONE
39 };
40
41 struct wreplsrv_out_connect_state {
42         enum wreplsrv_out_connect_stage stage;
43         struct composite_context *c;
44         struct wrepl_request *req;
45         struct composite_context *c_req;
46         struct wrepl_associate assoc_io;
47         enum winsrepl_partner_type type;
48         struct wreplsrv_out_connection *wreplconn;
49 };
50
51 static void wreplsrv_out_connect_handler_creq(struct composite_context *c_req);
52 static void wreplsrv_out_connect_handler_req(struct wrepl_request *req);
53
54 static NTSTATUS wreplsrv_out_connect_wait_socket(struct wreplsrv_out_connect_state *state)
55 {
56         NTSTATUS status;
57
58         status = wrepl_connect_recv(state->c_req);
59         NT_STATUS_NOT_OK_RETURN(status);
60
61         state->req = wrepl_associate_send(state->wreplconn->sock, &state->assoc_io);
62         NT_STATUS_HAVE_NO_MEMORY(state->req);
63
64         state->req->async.fn            = wreplsrv_out_connect_handler_req;
65         state->req->async.private       = state;
66
67         state->stage = WREPLSRV_OUT_CONNECT_STAGE_WAIT_ASSOC_CTX;
68
69         return NT_STATUS_OK;
70 }
71
72 static NTSTATUS wreplsrv_out_connect_wait_assoc_ctx(struct wreplsrv_out_connect_state *state)
73 {
74         NTSTATUS status;
75
76         status = wrepl_associate_recv(state->req, &state->assoc_io);
77         NT_STATUS_NOT_OK_RETURN(status);
78
79         state->wreplconn->assoc_ctx.peer_ctx = state->assoc_io.out.assoc_ctx;
80
81         if (state->type == WINSREPL_PARTNER_PUSH) {
82                 state->wreplconn->partner->push.wreplconn = state->wreplconn;
83                 talloc_steal(state->wreplconn->partner, state->wreplconn);
84         } else if (state->type == WINSREPL_PARTNER_PULL) {
85                 state->wreplconn->partner->pull.wreplconn = state->wreplconn;
86                 talloc_steal(state->wreplconn->partner, state->wreplconn);
87         }
88
89         state->stage = WREPLSRV_OUT_CONNECT_STAGE_DONE;
90
91         return NT_STATUS_OK;
92 }
93
94 static void wreplsrv_out_connect_handler(struct wreplsrv_out_connect_state *state)
95 {
96         struct composite_context *c = state->c;
97
98         switch (state->stage) {
99         case WREPLSRV_OUT_CONNECT_STAGE_WAIT_SOCKET:
100                 c->status = wreplsrv_out_connect_wait_socket(state);
101                 break;
102         case WREPLSRV_OUT_CONNECT_STAGE_WAIT_ASSOC_CTX:
103                 c->status = wreplsrv_out_connect_wait_assoc_ctx(state);
104                 c->state  = COMPOSITE_STATE_DONE;
105                 break;
106         case WREPLSRV_OUT_CONNECT_STAGE_DONE:
107                 c->status = NT_STATUS_INTERNAL_ERROR;
108         }
109
110         if (!NT_STATUS_IS_OK(c->status)) {
111                 c->state = COMPOSITE_STATE_ERROR;
112         }
113
114         if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
115                 c->async.fn(c);
116         }
117 }
118
119 static void wreplsrv_out_connect_handler_creq(struct composite_context *creq)
120 {
121         struct wreplsrv_out_connect_state *state = talloc_get_type(creq->async.private_data,
122                                                    struct wreplsrv_out_connect_state);
123         wreplsrv_out_connect_handler(state);
124         return;
125 }
126
127 static void wreplsrv_out_connect_handler_req(struct wrepl_request *req)
128 {
129         struct wreplsrv_out_connect_state *state = talloc_get_type(req->async.private,
130                                                    struct wreplsrv_out_connect_state);
131         wreplsrv_out_connect_handler(state);
132         return;
133 }
134
135 static struct composite_context *wreplsrv_out_connect_send(struct wreplsrv_partner *partner,
136                                                            enum winsrepl_partner_type type,
137                                                            struct wreplsrv_out_connection *wreplconn)
138 {
139         struct composite_context *c = NULL;
140         struct wreplsrv_service *service = partner->service;
141         struct wreplsrv_out_connect_state *state = NULL;
142         struct wreplsrv_out_connection **wreplconnp = &wreplconn;
143         bool cached_connection = false;
144
145         c = talloc_zero(partner, struct composite_context);
146         if (!c) goto failed;
147
148         state = talloc_zero(c, struct wreplsrv_out_connect_state);
149         if (!state) goto failed;
150         state->c        = c;
151         state->type     = type;
152
153         c->state        = COMPOSITE_STATE_IN_PROGRESS;
154         c->event_ctx    = service->task->event_ctx;
155         c->private_data = state;
156
157         if (type == WINSREPL_PARTNER_PUSH) {
158                 cached_connection       = true;
159                 wreplconn               = partner->push.wreplconn;
160                 wreplconnp              = &partner->push.wreplconn;
161         } else if (type == WINSREPL_PARTNER_PULL) {
162                 cached_connection       = true;
163                 wreplconn               = partner->pull.wreplconn;
164                 wreplconnp              = &partner->pull.wreplconn;
165         }
166
167         /* we have a connection already, so use it */
168         if (wreplconn) {
169                 if (!wreplconn->sock->dead) {
170                         state->stage    = WREPLSRV_OUT_CONNECT_STAGE_DONE;
171                         state->wreplconn= wreplconn;
172                         composite_done(c);
173                         return c;
174                 } else if (!cached_connection) {
175                         state->stage    = WREPLSRV_OUT_CONNECT_STAGE_DONE;
176                         state->wreplconn= NULL;
177                         composite_done(c);
178                         return c;
179                 } else {
180                         talloc_free(wreplconn);
181                         *wreplconnp = NULL;
182                 }
183         }
184
185         wreplconn = talloc_zero(state, struct wreplsrv_out_connection);
186         if (!wreplconn) goto failed;
187
188         wreplconn->service      = service;
189         wreplconn->partner      = partner;
190         wreplconn->sock         = wrepl_socket_init(wreplconn, service->task->event_ctx, lp_iconv_convenience(service->task->lp_ctx));
191         if (!wreplconn->sock) goto failed;
192
193         state->stage    = WREPLSRV_OUT_CONNECT_STAGE_WAIT_SOCKET;
194         state->wreplconn= wreplconn;
195         state->c_req    = wrepl_connect_send(wreplconn->sock,
196                                              lp_resolve_context(service->task->lp_ctx),
197                                              partner->our_address?partner->our_address:wrepl_best_ip(service->task->lp_ctx, partner->address),
198                                              partner->address);
199         if (!state->c_req) goto failed;
200
201         state->c_req->async.fn                  = wreplsrv_out_connect_handler_creq;
202         state->c_req->async.private_data        = state;
203
204         return c;
205 failed:
206         talloc_free(c);
207         return NULL;
208 }
209
210 static NTSTATUS wreplsrv_out_connect_recv(struct composite_context *c, TALLOC_CTX *mem_ctx,
211                                           struct wreplsrv_out_connection **wreplconn)
212 {
213         NTSTATUS status;
214
215         status = composite_wait(c);
216
217         if (NT_STATUS_IS_OK(status)) {
218                 struct wreplsrv_out_connect_state *state = talloc_get_type(c->private_data,
219                                                            struct wreplsrv_out_connect_state);
220                 if (state->wreplconn) {
221                         *wreplconn = talloc_reference(mem_ctx, state->wreplconn);
222                         if (!*wreplconn) status = NT_STATUS_NO_MEMORY;
223                 } else {
224                         status = NT_STATUS_INVALID_CONNECTION;
225                 }
226         }
227
228         talloc_free(c);
229         return status;
230         
231 }
232
233 struct wreplsrv_pull_table_io {
234         struct {
235                 struct wreplsrv_partner *partner;
236                 uint32_t num_owners;
237                 struct wrepl_wins_owner *owners;
238         } in;
239         struct {
240                 uint32_t num_owners;
241                 struct wrepl_wins_owner *owners;
242         } out;
243 };
244
245 enum wreplsrv_pull_table_stage {
246         WREPLSRV_PULL_TABLE_STAGE_WAIT_CONNECTION,
247         WREPLSRV_PULL_TABLE_STAGE_WAIT_TABLE_REPLY,
248         WREPLSRV_PULL_TABLE_STAGE_DONE
249 };
250
251 struct wreplsrv_pull_table_state {
252         enum wreplsrv_pull_table_stage stage;
253         struct composite_context *c;
254         struct wrepl_request *req;
255         struct wrepl_pull_table table_io;
256         struct wreplsrv_pull_table_io *io;
257         struct composite_context *creq;
258         struct wreplsrv_out_connection *wreplconn;
259 };
260
261 static void wreplsrv_pull_table_handler_req(struct wrepl_request *req);
262
263 static NTSTATUS wreplsrv_pull_table_wait_connection(struct wreplsrv_pull_table_state *state)
264 {
265         NTSTATUS status;
266
267         status = wreplsrv_out_connect_recv(state->creq, state, &state->wreplconn);
268         NT_STATUS_NOT_OK_RETURN(status);
269
270         state->table_io.in.assoc_ctx = state->wreplconn->assoc_ctx.peer_ctx;
271         state->req = wrepl_pull_table_send(state->wreplconn->sock, &state->table_io);
272         NT_STATUS_HAVE_NO_MEMORY(state->req);
273
274         state->req->async.fn            = wreplsrv_pull_table_handler_req;
275         state->req->async.private       = state;
276
277         state->stage = WREPLSRV_PULL_TABLE_STAGE_WAIT_TABLE_REPLY;
278
279         return NT_STATUS_OK;
280 }
281
282 static NTSTATUS wreplsrv_pull_table_wait_table_reply(struct wreplsrv_pull_table_state *state)
283 {
284         NTSTATUS status;
285
286         status = wrepl_pull_table_recv(state->req, state, &state->table_io);
287         NT_STATUS_NOT_OK_RETURN(status);
288
289         state->stage = WREPLSRV_PULL_TABLE_STAGE_DONE;
290
291         return NT_STATUS_OK;
292 }
293
294 static void wreplsrv_pull_table_handler(struct wreplsrv_pull_table_state *state)
295 {
296         struct composite_context *c = state->c;
297
298         switch (state->stage) {
299         case WREPLSRV_PULL_TABLE_STAGE_WAIT_CONNECTION:
300                 c->status = wreplsrv_pull_table_wait_connection(state);
301                 break;
302         case WREPLSRV_PULL_TABLE_STAGE_WAIT_TABLE_REPLY:
303                 c->status = wreplsrv_pull_table_wait_table_reply(state);
304                 c->state  = COMPOSITE_STATE_DONE;
305                 break;
306         case WREPLSRV_PULL_TABLE_STAGE_DONE:
307                 c->status = NT_STATUS_INTERNAL_ERROR;
308         }
309
310         if (!NT_STATUS_IS_OK(c->status)) {
311                 c->state = COMPOSITE_STATE_ERROR;
312         }
313
314         if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
315                 c->async.fn(c);
316         }
317 }
318
319 static void wreplsrv_pull_table_handler_creq(struct composite_context *creq)
320 {
321         struct wreplsrv_pull_table_state *state = talloc_get_type(creq->async.private_data,
322                                                   struct wreplsrv_pull_table_state);
323         wreplsrv_pull_table_handler(state);
324         return;
325 }
326
327 static void wreplsrv_pull_table_handler_req(struct wrepl_request *req)
328 {
329         struct wreplsrv_pull_table_state *state = talloc_get_type(req->async.private,
330                                                   struct wreplsrv_pull_table_state);
331         wreplsrv_pull_table_handler(state);
332         return;
333 }
334
335 static struct composite_context *wreplsrv_pull_table_send(TALLOC_CTX *mem_ctx, struct wreplsrv_pull_table_io *io)
336 {
337         struct composite_context *c = NULL;
338         struct wreplsrv_service *service = io->in.partner->service;
339         struct wreplsrv_pull_table_state *state = NULL;
340
341         c = talloc_zero(mem_ctx, struct composite_context);
342         if (!c) goto failed;
343
344         state = talloc_zero(c, struct wreplsrv_pull_table_state);
345         if (!state) goto failed;
346         state->c        = c;
347         state->io       = io;
348
349         c->state        = COMPOSITE_STATE_IN_PROGRESS;
350         c->event_ctx    = service->task->event_ctx;
351         c->private_data = state;
352
353         if (io->in.num_owners) {
354                 state->table_io.out.num_partners        = io->in.num_owners;
355                 state->table_io.out.partners            = io->in.owners;
356                 state->stage                            = WREPLSRV_PULL_TABLE_STAGE_DONE;
357                 composite_done(c);
358                 return c;
359         }
360
361         state->stage    = WREPLSRV_PULL_TABLE_STAGE_WAIT_CONNECTION;
362         state->creq     = wreplsrv_out_connect_send(io->in.partner, WINSREPL_PARTNER_PULL, NULL);
363         if (!state->creq) goto failed;
364
365         state->creq->async.fn           = wreplsrv_pull_table_handler_creq;
366         state->creq->async.private_data = state;
367
368         return c;
369 failed:
370         talloc_free(c);
371         return NULL;
372 }
373
374 static NTSTATUS wreplsrv_pull_table_recv(struct composite_context *c, TALLOC_CTX *mem_ctx,
375                                          struct wreplsrv_pull_table_io *io)
376 {
377         NTSTATUS status;
378
379         status = composite_wait(c);
380
381         if (NT_STATUS_IS_OK(status)) {
382                 struct wreplsrv_pull_table_state *state = talloc_get_type(c->private_data,
383                                                           struct wreplsrv_pull_table_state);
384                 io->out.num_owners      = state->table_io.out.num_partners;
385                 io->out.owners          = talloc_reference(mem_ctx, state->table_io.out.partners);
386         }
387
388         talloc_free(c);
389         return status;  
390 }
391
392 struct wreplsrv_pull_names_io {
393         struct {
394                 struct wreplsrv_partner *partner;
395                 struct wreplsrv_out_connection *wreplconn;
396                 struct wrepl_wins_owner owner;
397         } in;
398         struct {
399                 uint32_t num_names;
400                 struct wrepl_name *names;
401         } out;
402 };
403
404 enum wreplsrv_pull_names_stage {
405         WREPLSRV_PULL_NAMES_STAGE_WAIT_CONNECTION,
406         WREPLSRV_PULL_NAMES_STAGE_WAIT_SEND_REPLY,
407         WREPLSRV_PULL_NAMES_STAGE_DONE
408 };
409
410 struct wreplsrv_pull_names_state {
411         enum wreplsrv_pull_names_stage stage;
412         struct composite_context *c;
413         struct wrepl_request *req;
414         struct wrepl_pull_names pull_io;
415         struct wreplsrv_pull_names_io *io;
416         struct composite_context *creq;
417         struct wreplsrv_out_connection *wreplconn;
418 };
419
420 static void wreplsrv_pull_names_handler_req(struct wrepl_request *req);
421
422 static NTSTATUS wreplsrv_pull_names_wait_connection(struct wreplsrv_pull_names_state *state)
423 {
424         NTSTATUS status;
425
426         status = wreplsrv_out_connect_recv(state->creq, state, &state->wreplconn);
427         NT_STATUS_NOT_OK_RETURN(status);
428
429         state->pull_io.in.assoc_ctx     = state->wreplconn->assoc_ctx.peer_ctx;
430         state->pull_io.in.partner       = state->io->in.owner;
431         state->req = wrepl_pull_names_send(state->wreplconn->sock, &state->pull_io);
432         NT_STATUS_HAVE_NO_MEMORY(state->req);
433
434         state->req->async.fn            = wreplsrv_pull_names_handler_req;
435         state->req->async.private       = state;
436
437         state->stage = WREPLSRV_PULL_NAMES_STAGE_WAIT_SEND_REPLY;
438
439         return NT_STATUS_OK;
440 }
441
442 static NTSTATUS wreplsrv_pull_names_wait_send_reply(struct wreplsrv_pull_names_state *state)
443 {
444         NTSTATUS status;
445
446         status = wrepl_pull_names_recv(state->req, state, &state->pull_io);
447         NT_STATUS_NOT_OK_RETURN(status);
448
449         state->stage = WREPLSRV_PULL_NAMES_STAGE_DONE;
450
451         return NT_STATUS_OK;
452 }
453
454 static void wreplsrv_pull_names_handler(struct wreplsrv_pull_names_state *state)
455 {
456         struct composite_context *c = state->c;
457
458         switch (state->stage) {
459         case WREPLSRV_PULL_NAMES_STAGE_WAIT_CONNECTION:
460                 c->status = wreplsrv_pull_names_wait_connection(state);
461                 break;
462         case WREPLSRV_PULL_NAMES_STAGE_WAIT_SEND_REPLY:
463                 c->status = wreplsrv_pull_names_wait_send_reply(state);
464                 c->state  = COMPOSITE_STATE_DONE;
465                 break;
466         case WREPLSRV_PULL_NAMES_STAGE_DONE:
467                 c->status = NT_STATUS_INTERNAL_ERROR;
468         }
469
470         if (!NT_STATUS_IS_OK(c->status)) {
471                 c->state = COMPOSITE_STATE_ERROR;
472         }
473
474         if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
475                 c->async.fn(c);
476         }
477 }
478
479 static void wreplsrv_pull_names_handler_creq(struct composite_context *creq)
480 {
481         struct wreplsrv_pull_names_state *state = talloc_get_type(creq->async.private_data,
482                                                   struct wreplsrv_pull_names_state);
483         wreplsrv_pull_names_handler(state);
484         return;
485 }
486
487 static void wreplsrv_pull_names_handler_req(struct wrepl_request *req)
488 {
489         struct wreplsrv_pull_names_state *state = talloc_get_type(req->async.private,
490                                                   struct wreplsrv_pull_names_state);
491         wreplsrv_pull_names_handler(state);
492         return;
493 }
494
495 static struct composite_context *wreplsrv_pull_names_send(TALLOC_CTX *mem_ctx, struct wreplsrv_pull_names_io *io)
496 {
497         struct composite_context *c = NULL;
498         struct wreplsrv_service *service = io->in.partner->service;
499         struct wreplsrv_pull_names_state *state = NULL;
500         enum winsrepl_partner_type partner_type = WINSREPL_PARTNER_PULL;
501
502         if (io->in.wreplconn) partner_type = WINSREPL_PARTNER_NONE;
503
504         c = talloc_zero(mem_ctx, struct composite_context);
505         if (!c) goto failed;
506
507         state = talloc_zero(c, struct wreplsrv_pull_names_state);
508         if (!state) goto failed;
509         state->c        = c;
510         state->io       = io;
511
512         c->state        = COMPOSITE_STATE_IN_PROGRESS;
513         c->event_ctx    = service->task->event_ctx;
514         c->private_data = state;
515
516         state->stage    = WREPLSRV_PULL_NAMES_STAGE_WAIT_CONNECTION;
517         state->creq     = wreplsrv_out_connect_send(io->in.partner, partner_type, io->in.wreplconn);
518         if (!state->creq) goto failed;
519
520         state->creq->async.fn           = wreplsrv_pull_names_handler_creq;
521         state->creq->async.private_data = state;
522
523         return c;
524 failed:
525         talloc_free(c);
526         return NULL;
527 }
528
529 static NTSTATUS wreplsrv_pull_names_recv(struct composite_context *c, TALLOC_CTX *mem_ctx,
530                                          struct wreplsrv_pull_names_io *io)
531 {
532         NTSTATUS status;
533
534         status = composite_wait(c);
535
536         if (NT_STATUS_IS_OK(status)) {
537                 struct wreplsrv_pull_names_state *state = talloc_get_type(c->private_data,
538                                                           struct wreplsrv_pull_names_state);
539                 io->out.num_names       = state->pull_io.out.num_names;
540                 io->out.names           = talloc_reference(mem_ctx, state->pull_io.out.names);
541         }
542
543         talloc_free(c);
544         return status;
545         
546 }
547
548 enum wreplsrv_pull_cycle_stage {
549         WREPLSRV_PULL_CYCLE_STAGE_WAIT_TABLE_REPLY,
550         WREPLSRV_PULL_CYCLE_STAGE_WAIT_SEND_REPLIES,
551         WREPLSRV_PULL_CYCLE_STAGE_WAIT_STOP_ASSOC,
552         WREPLSRV_PULL_CYCLE_STAGE_DONE
553 };
554
555 struct wreplsrv_pull_cycle_state {
556         enum wreplsrv_pull_cycle_stage stage;
557         struct composite_context *c;
558         struct wreplsrv_pull_cycle_io *io;
559         struct wreplsrv_pull_table_io table_io;
560         uint32_t current;
561         struct wreplsrv_pull_names_io names_io;
562         struct composite_context *creq;
563         struct wrepl_associate_stop assoc_stop_io;
564         struct wrepl_request *req;
565 };
566
567 static void wreplsrv_pull_cycle_handler_creq(struct composite_context *creq);
568 static void wreplsrv_pull_cycle_handler_req(struct wrepl_request *req);
569
570 static NTSTATUS wreplsrv_pull_cycle_next_owner_do_work(struct wreplsrv_pull_cycle_state *state)
571 {
572         struct wreplsrv_owner *current_owner=NULL;
573         struct wreplsrv_owner *local_owner;
574         uint32_t i;
575         uint64_t old_max_version = 0;
576         bool do_pull = false;
577
578         for (i=state->current; i < state->table_io.out.num_owners; i++) {
579                 current_owner = wreplsrv_find_owner(state->io->in.partner->service,
580                                                     state->io->in.partner->pull.table,
581                                                     state->table_io.out.owners[i].address);
582
583                 local_owner = wreplsrv_find_owner(state->io->in.partner->service,
584                                                   state->io->in.partner->service->table,
585                                                   state->table_io.out.owners[i].address);
586                 /*
587                  * this means we are ourself the current owner,
588                  * and we don't want replicate ourself
589                  */
590                 if (!current_owner) continue;
591
592                 /*
593                  * this means we don't have any records of this owner
594                  * so fetch them
595                  */
596                 if (!local_owner) {
597                         do_pull         = true;
598                         
599                         break;
600                 }
601
602                 /*
603                  * this means the remote partner has some new records of this owner
604                  * fetch them
605                  */
606                 if (current_owner->owner.max_version > local_owner->owner.max_version) {
607                         do_pull         = true;
608                         old_max_version = local_owner->owner.max_version;
609                         break;
610                 }
611         }
612         state->current = i;
613
614         if (do_pull) {
615                 state->names_io.in.partner              = state->io->in.partner;
616                 state->names_io.in.wreplconn            = state->io->in.wreplconn;
617                 state->names_io.in.owner                = current_owner->owner;
618                 state->names_io.in.owner.min_version    = old_max_version + 1;
619                 state->creq = wreplsrv_pull_names_send(state, &state->names_io);
620                 NT_STATUS_HAVE_NO_MEMORY(state->creq);
621
622                 state->creq->async.fn           = wreplsrv_pull_cycle_handler_creq;
623                 state->creq->async.private_data = state;
624
625                 return STATUS_MORE_ENTRIES;
626         }
627
628         return NT_STATUS_OK;
629 }
630
631 static NTSTATUS wreplsrv_pull_cycle_next_owner_wrapper(struct wreplsrv_pull_cycle_state *state)
632 {
633         NTSTATUS status;
634
635         status = wreplsrv_pull_cycle_next_owner_do_work(state);
636         if (NT_STATUS_IS_OK(status)) {
637                 state->stage = WREPLSRV_PULL_CYCLE_STAGE_DONE;
638         } else if (NT_STATUS_EQUAL(STATUS_MORE_ENTRIES, status)) {
639                 state->stage = WREPLSRV_PULL_CYCLE_STAGE_WAIT_SEND_REPLIES;
640                 status = NT_STATUS_OK;
641         }
642
643         if (state->stage == WREPLSRV_PULL_CYCLE_STAGE_DONE && state->io->in.wreplconn) {
644                 state->assoc_stop_io.in.assoc_ctx       = state->io->in.wreplconn->assoc_ctx.peer_ctx;
645                 state->assoc_stop_io.in.reason          = 0;
646                 state->req = wrepl_associate_stop_send(state->io->in.wreplconn->sock, &state->assoc_stop_io);
647                 NT_STATUS_HAVE_NO_MEMORY(state->req);
648
649                 state->req->async.fn            = wreplsrv_pull_cycle_handler_req;
650                 state->req->async.private       = state;
651
652                 state->stage = WREPLSRV_PULL_CYCLE_STAGE_WAIT_STOP_ASSOC;
653         }
654
655         return status;
656 }
657
658 static NTSTATUS wreplsrv_pull_cycle_wait_table_reply(struct wreplsrv_pull_cycle_state *state)
659 {
660         NTSTATUS status;
661         uint32_t i;
662
663         status = wreplsrv_pull_table_recv(state->creq, state, &state->table_io);
664         NT_STATUS_NOT_OK_RETURN(status);
665
666         /* update partner table */
667         for (i=0; i < state->table_io.out.num_owners; i++) {
668                 status = wreplsrv_add_table(state->io->in.partner->service,
669                                             state->io->in.partner, 
670                                             &state->io->in.partner->pull.table,
671                                             state->table_io.out.owners[i].address,
672                                             state->table_io.out.owners[i].max_version);
673                 NT_STATUS_NOT_OK_RETURN(status);
674         }
675
676         status = wreplsrv_pull_cycle_next_owner_wrapper(state);
677         NT_STATUS_NOT_OK_RETURN(status);
678
679         return status;
680 }
681
682 static NTSTATUS wreplsrv_pull_cycle_apply_records(struct wreplsrv_pull_cycle_state *state)
683 {
684         NTSTATUS status;
685
686         status = wreplsrv_apply_records(state->io->in.partner,
687                                         &state->names_io.in.owner,
688                                         state->names_io.out.num_names,
689                                         state->names_io.out.names);
690         NT_STATUS_NOT_OK_RETURN(status);
691
692         talloc_free(state->names_io.out.names);
693         ZERO_STRUCT(state->names_io);
694
695         return NT_STATUS_OK;
696 }
697
698 static NTSTATUS wreplsrv_pull_cycle_wait_send_replies(struct wreplsrv_pull_cycle_state *state)
699 {
700         NTSTATUS status;
701
702         status = wreplsrv_pull_names_recv(state->creq, state, &state->names_io);
703         NT_STATUS_NOT_OK_RETURN(status);
704
705         /*
706          * TODO: this should maybe an async call,
707          *       because we may need some network access
708          *       for conflict resolving
709          */
710         status = wreplsrv_pull_cycle_apply_records(state);
711         NT_STATUS_NOT_OK_RETURN(status);
712
713         status = wreplsrv_pull_cycle_next_owner_wrapper(state);
714         NT_STATUS_NOT_OK_RETURN(status);
715
716         return status;
717 }
718
719 static NTSTATUS wreplsrv_pull_cycle_wait_stop_assoc(struct wreplsrv_pull_cycle_state *state)
720 {
721         NTSTATUS status;
722
723         status = wrepl_associate_stop_recv(state->req, &state->assoc_stop_io);
724         NT_STATUS_NOT_OK_RETURN(status);
725
726         state->stage = WREPLSRV_PULL_CYCLE_STAGE_DONE;
727
728         return status;
729 }
730
731 static void wreplsrv_pull_cycle_handler(struct wreplsrv_pull_cycle_state *state)
732 {
733         struct composite_context *c = state->c;
734
735         switch (state->stage) {
736         case WREPLSRV_PULL_CYCLE_STAGE_WAIT_TABLE_REPLY:
737                 c->status = wreplsrv_pull_cycle_wait_table_reply(state);
738                 break;
739         case WREPLSRV_PULL_CYCLE_STAGE_WAIT_SEND_REPLIES:
740                 c->status = wreplsrv_pull_cycle_wait_send_replies(state);
741                 break;
742         case WREPLSRV_PULL_CYCLE_STAGE_WAIT_STOP_ASSOC:
743                 c->status = wreplsrv_pull_cycle_wait_stop_assoc(state);
744                 break;
745         case WREPLSRV_PULL_CYCLE_STAGE_DONE:
746                 c->status = NT_STATUS_INTERNAL_ERROR;
747         }
748
749         if (state->stage == WREPLSRV_PULL_CYCLE_STAGE_DONE) {
750                 c->state  = COMPOSITE_STATE_DONE;
751         }
752
753         if (!NT_STATUS_IS_OK(c->status)) {
754                 c->state = COMPOSITE_STATE_ERROR;
755         }
756
757         if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
758                 c->async.fn(c);
759         }
760 }
761
762 static void wreplsrv_pull_cycle_handler_creq(struct composite_context *creq)
763 {
764         struct wreplsrv_pull_cycle_state *state = talloc_get_type(creq->async.private_data,
765                                                   struct wreplsrv_pull_cycle_state);
766         wreplsrv_pull_cycle_handler(state);
767         return;
768 }
769
770 static void wreplsrv_pull_cycle_handler_req(struct wrepl_request *req)
771 {
772         struct wreplsrv_pull_cycle_state *state = talloc_get_type(req->async.private,
773                                                   struct wreplsrv_pull_cycle_state);
774         wreplsrv_pull_cycle_handler(state);
775         return;
776 }
777
778 struct composite_context *wreplsrv_pull_cycle_send(TALLOC_CTX *mem_ctx, struct wreplsrv_pull_cycle_io *io)
779 {
780         struct composite_context *c = NULL;
781         struct wreplsrv_service *service = io->in.partner->service;
782         struct wreplsrv_pull_cycle_state *state = NULL;
783
784         c = talloc_zero(mem_ctx, struct composite_context);
785         if (!c) goto failed;
786
787         state = talloc_zero(c, struct wreplsrv_pull_cycle_state);
788         if (!state) goto failed;
789         state->c        = c;
790         state->io       = io;
791
792         c->state        = COMPOSITE_STATE_IN_PROGRESS;
793         c->event_ctx    = service->task->event_ctx;
794         c->private_data = state;
795
796         state->stage    = WREPLSRV_PULL_CYCLE_STAGE_WAIT_TABLE_REPLY;
797         state->table_io.in.partner      = io->in.partner;
798         state->table_io.in.num_owners   = io->in.num_owners;
799         state->table_io.in.owners       = io->in.owners;
800         state->creq = wreplsrv_pull_table_send(state, &state->table_io);
801         if (!state->creq) goto failed;
802
803         state->creq->async.fn           = wreplsrv_pull_cycle_handler_creq;
804         state->creq->async.private_data = state;
805
806         return c;
807 failed:
808         talloc_free(c);
809         return NULL;
810 }
811
812 NTSTATUS wreplsrv_pull_cycle_recv(struct composite_context *c)
813 {
814         NTSTATUS status;
815
816         status = composite_wait(c);
817
818         talloc_free(c);
819         return status;
820 }
821
822 enum wreplsrv_push_notify_stage {
823         WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_CONNECT,
824         WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_INFORM,
825         WREPLSRV_PUSH_NOTIFY_STAGE_DONE
826 };
827
828 struct wreplsrv_push_notify_state {
829         enum wreplsrv_push_notify_stage stage;
830         struct composite_context *c;
831         struct wreplsrv_push_notify_io *io;
832         enum wrepl_replication_cmd command;
833         bool full_table;
834         struct wrepl_send_ctrl ctrl;
835         struct wrepl_request *req;
836         struct wrepl_packet req_packet;
837         struct wrepl_packet *rep_packet;
838         struct composite_context *creq;
839         struct wreplsrv_out_connection *wreplconn;
840 };
841
842 static void wreplsrv_push_notify_handler_creq(struct composite_context *creq);
843 static void wreplsrv_push_notify_handler_req(struct wrepl_request *req);
844
845 static NTSTATUS wreplsrv_push_notify_update(struct wreplsrv_push_notify_state *state)
846 {
847         struct wreplsrv_service *service = state->io->in.partner->service;
848         struct wrepl_packet *req = &state->req_packet;
849         struct wrepl_replication *repl_out = &state->req_packet.message.replication;
850         struct wrepl_table *table_out = &state->req_packet.message.replication.info.table;
851         struct wreplsrv_in_connection *wrepl_in;
852         NTSTATUS status;
853         struct socket_context *sock;
854         struct packet_context *packet;
855         uint16_t fde_flags;
856
857         /* prepare the outgoing request */
858         req->opcode     = WREPL_OPCODE_BITS;
859         req->assoc_ctx  = state->wreplconn->assoc_ctx.peer_ctx;
860         req->mess_type  = WREPL_REPLICATION;
861
862         repl_out->command = state->command;
863
864         status = wreplsrv_fill_wrepl_table(service, state, table_out,
865                                            service->wins_db->local_owner, state->full_table);
866         NT_STATUS_NOT_OK_RETURN(status);
867
868         /* queue the request */
869         state->req = wrepl_request_send(state->wreplconn->sock, req, NULL);
870         NT_STATUS_HAVE_NO_MEMORY(state->req);
871
872         /*
873          * now we need to convert the wrepl_socket (client connection)
874          * into a wreplsrv_in_connection (server connection), because
875          * we'll act as a server on this connection after the WREPL_REPL_UPDATE*
876          * message is received by the peer.
877          */
878
879         /* steal the socket_context */
880         sock = state->wreplconn->sock->sock;
881         state->wreplconn->sock->sock = NULL;
882         talloc_steal(state, sock);
883
884         /* 
885          * steal the packet_context
886          * note the request DATA_BLOB we just send on the
887          * wrepl_socket (client connection) is still unter the 
888          * packet context and will be send to the wire
889          */
890         packet = state->wreplconn->sock->packet;
891         state->wreplconn->sock->packet = NULL;
892         talloc_steal(state, packet);
893
894         /*
895          * get the fde_flags of the old fde event,
896          * so that we can later set the same flags to the new one
897          */
898         fde_flags = event_get_fd_flags(state->wreplconn->sock->event.fde);
899
900         /*
901          * free the wrepl_socket (client connection)
902          */
903         talloc_free(state->wreplconn->sock);
904         state->wreplconn->sock = NULL;
905
906         /*
907          * now create a wreplsrv_in_connection,
908          * on which we act as server
909          *
910          * NOTE: sock and packet will be stolen by
911          *       wreplsrv_in_connection_merge()
912          */
913         status = wreplsrv_in_connection_merge(state->io->in.partner,
914                                               sock, packet, &wrepl_in);
915         NT_STATUS_NOT_OK_RETURN(status);
916
917         event_set_fd_flags(wrepl_in->conn->event.fde, fde_flags);
918
919         wrepl_in->assoc_ctx.peer_ctx    = state->wreplconn->assoc_ctx.peer_ctx;
920         wrepl_in->assoc_ctx.our_ctx     = 0;
921
922         /* now we can free the wreplsrv_out_connection */
923         talloc_free(state->wreplconn);
924         state->wreplconn = NULL;
925
926         state->stage = WREPLSRV_PUSH_NOTIFY_STAGE_DONE;
927
928         return NT_STATUS_OK;
929 }
930
931 static NTSTATUS wreplsrv_push_notify_inform(struct wreplsrv_push_notify_state *state)
932 {
933         struct wreplsrv_service *service = state->io->in.partner->service;
934         struct wrepl_packet *req = &state->req_packet;
935         struct wrepl_replication *repl_out = &state->req_packet.message.replication;
936         struct wrepl_table *table_out = &state->req_packet.message.replication.info.table;
937         NTSTATUS status;
938
939         req->opcode     = WREPL_OPCODE_BITS;
940         req->assoc_ctx  = state->wreplconn->assoc_ctx.peer_ctx;
941         req->mess_type  = WREPL_REPLICATION;
942
943         repl_out->command = state->command;
944
945         status = wreplsrv_fill_wrepl_table(service, state, table_out,
946                                            service->wins_db->local_owner, state->full_table);
947         NT_STATUS_NOT_OK_RETURN(status);
948
949         /* we won't get a reply to a inform message */
950         state->ctrl.send_only           = true;
951
952         state->req = wrepl_request_send(state->wreplconn->sock, req, &state->ctrl);
953         NT_STATUS_HAVE_NO_MEMORY(state->req);
954
955         state->req->async.fn            = wreplsrv_push_notify_handler_req;
956         state->req->async.private       = state;
957
958         state->stage = WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_INFORM;
959
960         return NT_STATUS_OK;
961 }
962
963 static NTSTATUS wreplsrv_push_notify_wait_connect(struct wreplsrv_push_notify_state *state)
964 {
965         NTSTATUS status;
966
967         status = wreplsrv_out_connect_recv(state->creq, state, &state->wreplconn);
968         NT_STATUS_NOT_OK_RETURN(status);
969
970         switch (state->command) {
971         case WREPL_REPL_UPDATE:
972                 state->full_table = true;
973                 return wreplsrv_push_notify_update(state);
974         case WREPL_REPL_UPDATE2:
975                 state->full_table = false;
976                 return wreplsrv_push_notify_update(state);
977         case WREPL_REPL_INFORM:
978                 state->full_table = true;
979                 return wreplsrv_push_notify_inform(state);
980         case WREPL_REPL_INFORM2:
981                 state->full_table = false;
982                 return wreplsrv_push_notify_inform(state);
983         default:
984                 return NT_STATUS_INTERNAL_ERROR;
985         }
986
987         return NT_STATUS_INTERNAL_ERROR;
988 }
989
990 static NTSTATUS wreplsrv_push_notify_wait_inform(struct wreplsrv_push_notify_state *state)
991 {
992         NTSTATUS status;
993
994         status =  wrepl_request_recv(state->req, state, NULL);
995         NT_STATUS_NOT_OK_RETURN(status);
996
997         state->stage = WREPLSRV_PUSH_NOTIFY_STAGE_DONE;
998         return status;
999 }
1000
1001 static void wreplsrv_push_notify_handler(struct wreplsrv_push_notify_state *state)
1002 {
1003         struct composite_context *c = state->c;
1004
1005         switch (state->stage) {
1006         case WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_CONNECT:
1007                 c->status = wreplsrv_push_notify_wait_connect(state);
1008                 break;
1009         case WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_INFORM:
1010                 c->status = wreplsrv_push_notify_wait_inform(state);
1011                 break;
1012         case WREPLSRV_PUSH_NOTIFY_STAGE_DONE:
1013                 c->status = NT_STATUS_INTERNAL_ERROR;
1014         }
1015
1016         if (state->stage == WREPLSRV_PUSH_NOTIFY_STAGE_DONE) {
1017                 c->state  = COMPOSITE_STATE_DONE;
1018         }
1019
1020         if (!NT_STATUS_IS_OK(c->status)) {
1021                 c->state = COMPOSITE_STATE_ERROR;
1022         }
1023
1024         if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
1025                 c->async.fn(c);
1026         }
1027 }
1028
1029 static void wreplsrv_push_notify_handler_creq(struct composite_context *creq)
1030 {
1031         struct wreplsrv_push_notify_state *state = talloc_get_type(creq->async.private_data,
1032                                                    struct wreplsrv_push_notify_state);
1033         wreplsrv_push_notify_handler(state);
1034         return;
1035 }
1036
1037 static void wreplsrv_push_notify_handler_req(struct wrepl_request *req)
1038 {
1039         struct wreplsrv_push_notify_state *state = talloc_get_type(req->async.private,
1040                                                    struct wreplsrv_push_notify_state);
1041         wreplsrv_push_notify_handler(state);
1042         return;
1043 }
1044
1045 struct composite_context *wreplsrv_push_notify_send(TALLOC_CTX *mem_ctx, struct wreplsrv_push_notify_io *io)
1046 {
1047         struct composite_context *c = NULL;
1048         struct wreplsrv_service *service = io->in.partner->service;
1049         struct wreplsrv_push_notify_state *state = NULL;
1050         enum winsrepl_partner_type partner_type;
1051
1052         c = talloc_zero(mem_ctx, struct composite_context);
1053         if (!c) goto failed;
1054
1055         state = talloc_zero(c, struct wreplsrv_push_notify_state);
1056         if (!state) goto failed;
1057         state->c        = c;
1058         state->io       = io;
1059
1060         if (io->in.inform) {
1061                 /* we can cache the connection in partner->push->wreplconn */
1062                 partner_type = WINSREPL_PARTNER_PUSH;
1063                 if (io->in.propagate) {
1064                         state->command  = WREPL_REPL_INFORM2;
1065                 } else {
1066                         state->command  = WREPL_REPL_INFORM;
1067                 }
1068         } else {
1069                 /* we can NOT cache the connection */
1070                 partner_type = WINSREPL_PARTNER_NONE;
1071                 if (io->in.propagate) {
1072                         state->command  = WREPL_REPL_UPDATE2;
1073                 } else {
1074                         state->command  = WREPL_REPL_UPDATE;
1075                 }       
1076         }
1077
1078         c->state        = COMPOSITE_STATE_IN_PROGRESS;
1079         c->event_ctx    = service->task->event_ctx;
1080         c->private_data = state;
1081
1082         state->stage    = WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_CONNECT;
1083         state->creq     = wreplsrv_out_connect_send(io->in.partner, partner_type, NULL);
1084         if (!state->creq) goto failed;
1085
1086         state->creq->async.fn           = wreplsrv_push_notify_handler_creq;
1087         state->creq->async.private_data = state;
1088
1089         return c;
1090 failed:
1091         talloc_free(c);
1092         return NULL;
1093 }
1094
1095 NTSTATUS wreplsrv_push_notify_recv(struct composite_context *c)
1096 {
1097         NTSTATUS status;
1098
1099         status = composite_wait(c);
1100
1101         talloc_free(c);
1102         return status;
1103 }